From f3a8f55f112269390d3fb8dfb7a1f46ca9199875 Mon Sep 17 00:00:00 2001 From: Roland Walker Date: Sat, 4 Jul 2026 08:47:42 -0400 Subject: [PATCH] add a /dsn command to manage DSN aliases The logic is almost exactly like favorite queries, but the interface is modernized to follow the subcommand pattern. Whereas favorite queries are accessed through several different commands, all saved-DSN functionality is via subcommands of /dsn, such as "/dsn list". --- changelog.md | 8 ++ mycli/TIPS | 2 + mycli/client.py | 2 + mycli/packages/special/dsn_aliases.py | 60 ++++++++++ mycli/packages/special/iocommands.py | 48 +++++++- mycli/packages/special/utils.py | 22 ++++ test/features/fixture_data/help_commands.txt | 77 ++++++------- test/pytests/test_dsn_aliases.py | 102 +++++++++++++++++ ...est_smart_completion_public_schema_only.py | 5 +- test/pytests/test_special_iocommands.py | 104 ++++++++++++++++++ test/pytests/test_special_utils.py | 44 ++++++++ 11 files changed, 434 insertions(+), 40 deletions(-) create mode 100644 mycli/packages/special/dsn_aliases.py create mode 100644 test/pytests/test_dsn_aliases.py diff --git a/changelog.md b/changelog.md index fba24e91..0012235b 100644 --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,11 @@ +Upcoming (TBD) +============== + +Features +--------- +* Add a `/dsn` command to manage persisted DSN aliases. + + 2.0.0 (2026/07/03) ============== diff --git a/mycli/TIPS b/mycli/TIPS index 4cf52077..a0f45761 100644 --- a/mycli/TIPS +++ b/mycli/TIPS @@ -108,6 +108,8 @@ the /watch command executes a query every N seconds! use /bug to file a bug on GitHub! +use /dsn to manage saved DSNs! + ### ### environment variables ### diff --git a/mycli/client.py b/mycli/client.py index 320d2fa8..bfccdbfc 100644 --- a/mycli/client.py +++ b/mycli/client.py @@ -35,6 +35,7 @@ from mycli.main_modes import repl as repl_package from mycli.output import OutputMixin from mycli.packages import special +from mycli.packages.special.dsn_aliases import DsnAliases from mycli.packages.special.favoritequeries import FavoriteQueries from mycli.packages.tabular_output import sql_format from mycli.schema_prefetcher import SchemaPrefetcher @@ -106,6 +107,7 @@ def __init__( self.default_keepalive_ticks = c['connection'].as_int('default_keepalive_ticks') FavoriteQueries.instance = FavoriteQueries.from_config(self.config) + DsnAliases.instance = DsnAliases.from_config(self.config) self.dsn_alias: str | None = None self.main_formatter = TabularOutputFormatter(format_name=c["main"]["table_format"]) diff --git a/mycli/packages/special/dsn_aliases.py b/mycli/packages/special/dsn_aliases.py new file mode 100644 index 00000000..07b93c15 --- /dev/null +++ b/mycli/packages/special/dsn_aliases.py @@ -0,0 +1,60 @@ +from __future__ import annotations + + +class DsnAliases: + section_name: str = 'alias_dsn' + + usage = """ +DSN aliases are a way to save frequently used connections +with a short alias. You can manage them by editing ~/.myclirc +directly or by using this command. + +Examples: + + # List all DSN aliases + mysql> /dsn list + ┌───────┬───────────────────────────────┐ + │ Alias │ DSN │ + ├───────┼───────────────────────────────┤ + │ rocks │ mysql://mycli@localhost/mysql │ + └───────┴───────────────────────────────┘ + + # Save a new DSN alias based on the current connection. + # The password will not be included! + mysql> /dsn save connection_1 + + # Delete a DSN alias. + mysql> /dsn delete connection_1 +""" + + # Class-level variable, for convenience to use as a singleton. + instance: DsnAliases + + def __init__(self, config) -> None: + self.config = config + + @classmethod + def from_config(cls, config): + return DsnAliases(config) + + def list(self) -> list[str]: + return list(self.config.get(self.section_name, {})) + + def get(self, alias: str) -> str | None: + return self.config.get(self.section_name, {}).get(alias, None) + + def save(self, alias: str, dsn: str) -> str: + self.config.encoding = 'utf-8' + if self.section_name not in self.config: + self.config[self.section_name] = {} + self.config[self.section_name][alias] = dsn + self.config.write() + return f'Saved: {alias}' + + def delete(self, alias: str) -> str: + try: + del self.config[self.section_name][alias] + except KeyError: + return f'Not Found: {alias}' + self.config.write() + return f'Deleted: {alias}' diff --git a/mycli/packages/special/iocommands.py b/mycli/packages/special/iocommands.py index 3b7effd9..1dd74374 100644 --- a/mycli/packages/special/iocommands.py +++ b/mycli/packages/special/iocommands.py @@ -19,11 +19,12 @@ from mycli.compat import WIN from mycli.packages.interactive_utils import confirm_destructive_query from mycli.packages.special.delimitercommand import DelimiterCommand +from mycli.packages.special.dsn_aliases import DsnAliases from mycli.packages.special.favoritequeries import FavoriteQueries from mycli.packages.special.main import COMMANDS as SPECIAL_COMMANDS from mycli.packages.special.main import ArgType, SpecialCommandAlias, special_command from mycli.packages.special.main import execute as special_execute -from mycli.packages.special.utils import handle_cd_command +from mycli.packages.special.utils import compute_current_dsn, handle_cd_command from mycli.packages.sqlresult import SQLResult sqlparse.engine.grouping.MAX_GROUPING_DEPTH = None # type: ignore[assignment] @@ -45,6 +46,7 @@ } delimiter_command = DelimiterCommand() favoritequeries = FavoriteQueries(ConfigObj()) +dsn_aliases = DsnAliases(ConfigObj()) DESTRUCTIVE_KEYWORDS: list[str] = [] SHOW_WARNINGS_ENABLED: bool = False @@ -439,6 +441,50 @@ def delete_favorite_query(arg: str, **_) -> list[SQLResult]: return [SQLResult(status=status)] +@special_command( + r'\dsn', + '/dsn ', + 'Manage saved DSNs.', + arg_type=ArgType.PARSED_QUERY, + case_sensitive=False, +) +def dsn( + cur: Cursor, + arg: str | None = None, + arg_type: ArgType = ArgType.PARSED_QUERY, + command_verbosity: bool = False, +) -> list[SQLResult]: + args = shlex.split(arg or '') + if len(args) == 1 and args[0].lower() == 'show': + dsn = compute_current_dsn(cur) + header = ['Current Connection'] + show_rows = [(dsn,)] + return [SQLResult(header=header, rows=show_rows)] + elif args and args[0].lower() == 'save': + if len(args) != 2: + return [SQLResult(status='Error: a single alias-name argument is required to save.')] + dsn = compute_current_dsn(cur) + alias = args[1] + status = DsnAliases.instance.save(alias, dsn) + return [SQLResult(status=status)] + elif args and args[0].lower() == 'delete': + if len(args) != 2: + return [SQLResult(status='Error: a single alias-name argument is required to delete.')] + alias = args[1] + status = DsnAliases.instance.delete(alias) + return [SQLResult(status=status)] + elif args and args[0].lower() == 'list': + header = ['Alias', 'DSN'] + list_rows = [(r, DsnAliases.instance.get(r)) for r in DsnAliases.instance.list()] + if not list_rows: + status = 'No DSN Aliases found.' + else: + status = None + return [SQLResult(status=status, header=header, rows=list_rows)] + else: + return [SQLResult(preamble=DsnAliases.instance.usage)] + + @special_command( "system", "/system [-r] ", diff --git a/mycli/packages/special/utils.py b/mycli/packages/special/utils.py index fc014323..a7342eb6 100644 --- a/mycli/packages/special/utils.py +++ b/mycli/packages/special/utils.py @@ -2,6 +2,8 @@ import logging import os from typing import Any +from urllib.parse import quote as urlquote +from urllib.parse import urlencode import click import pymysql @@ -143,3 +145,23 @@ def get_server_timezone(variables: dict[str, Any]) -> str: def get_local_timezone() -> str: return datetime.datetime.now().astimezone().tzname() or '' + + +def compute_current_dsn(cur: Cursor) -> str: + conn = cur.connection + user = urlquote(conn.user or '') + host = conn.host or 'localhost' + port = f':{conn.port}' + db = urlquote(conn.db.decode() if isinstance(conn.db, bytes) else (conn.db or '')) + if db: + db = f'/{db}' + query_part = {} + if getattr(conn, 'unix_socket', None): + query_part['socket'] = conn.unix_socket + port = '' + if getattr(conn, 'charset', None) and conn.charset != 'utf8mb4': + query_part['character_set'] = conn.charset + dsn = f'mysql://{user}@{host}{port}{db}' + if query_part: + dsn += '?' + urlencode(query_part) + return dsn diff --git a/test/features/fixture_data/help_commands.txt b/test/features/fixture_data/help_commands.txt index f3a25601..408125d1 100644 --- a/test/features/fixture_data/help_commands.txt +++ b/test/features/fixture_data/help_commands.txt @@ -1,38 +1,39 @@ -+-----------------+----------+---------------------------------+-------------------------------------------------------------+ -| Command | Shortcut | Usage | Description | -+-----------------+----------+---------------------------------+-------------------------------------------------------------+ -| /bug | | /bug | File a bug on GitHub. | -| /clip | | /clip | \clip | Copy query to the system clipboard. | -| /dt | | /dt[+] [table] | List or describe tables. | -| /edit | /e | /edit | \edit | Edit query with editor (uses $VISUAL or $EDITOR). | -| /f | | /f [name [args..]] | List or execute favorite queries. | -| /fd | | /fd | Delete a favorite query. | -| /fs | | /fs | Save a favorite query. | -| \g | | \g | Display query results (mnemonic: go). | -| \G | | \G | Display query results vertically. | -| /l | | /l | List databases. | -| /llm | /ai | /llm [arguments] | Interrogate an LLM. See "/llm help". | -| /once | /o | /once [-o] | Append next result to an output file (overwrite using -o). | -| /pipe_once | /| | /pipe_once | Send next result to a subprocess. | -| /timing | /t | /timing | Toggle timing of queries. | -| /connect | /r | /connect [database] | Reconnect to the server, optionally switching databases. | -| /delimiter | | /delimiter | Change end-of-statement delimiter. | -| /exit | /q | /exit | Exit. | -| /help | /? | /help [term] | Show this table, or search for help on a term. | -| /nopager | /n | /nopager | Disable pager; print to stdout. | -| /notee | | /notee | Stop writing results to an output file. | -| /nowarnings | /w | /nowarnings | Disable automatic warnings display. | -| /pager | /P | /pager [command] | Set pager to [command]. Print query results via pager. | -| /prompt | /R | /prompt | Change prompt format. | -| /quit | /q | /quit | Quit. | -| /redirectformat | /Tr | /redirectformat | Change the table format used to output redirected results. | -| /rehash | /# | /rehash | Refresh auto-completions. | -| /source | /. | /source | Execute queries from a file. | -| /status | /s | /status | Get status information from the server. | -| /system | | /system [-r] | Execute a system shell command (raw mode with -r). | -| /tableformat | /T | /tableformat | Change the table format used to output interactive results. | -| /tee | | /tee [-o] | Append all results to an output file (overwrite using -o). | -| /use | /u | /use | Change to a new database. | -| /warnings | /W | /warnings | Enable automatic warnings display. | -| /watch | | /watch [seconds] [-c] | Execute query every [seconds] seconds (5 by default). | -+-----------------+----------+---------------------------------+-------------------------------------------------------------+ ++-----------------+----------+-----------------------------------+-------------------------------------------------------------+ +| Command | Shortcut | Usage | Description | ++-----------------+----------+-----------------------------------+-------------------------------------------------------------+ +| /bug | | /bug | File a bug on GitHub. | +| /clip | | /clip | \clip | Copy query to the system clipboard. | +| /dsn | | /dsn | Manage saved DSNs. | +| /dt | | /dt[+] [table] | List or describe tables. | +| /edit | /e | /edit | \edit | Edit query with editor (uses $VISUAL or $EDITOR). | +| /f | | /f [name [args..]] | List or execute favorite queries. | +| /fd | | /fd | Delete a favorite query. | +| /fs | | /fs | Save a favorite query. | +| \g | | \g | Display query results (mnemonic: go). | +| \G | | \G | Display query results vertically. | +| /l | | /l | List databases. | +| /llm | /ai | /llm [arguments] | Interrogate an LLM. See "/llm help". | +| /once | /o | /once [-o] | Append next result to an output file (overwrite using -o). | +| /pipe_once | /| | /pipe_once | Send next result to a subprocess. | +| /timing | /t | /timing | Toggle timing of queries. | +| /connect | /r | /connect [database] | Reconnect to the server, optionally switching databases. | +| /delimiter | | /delimiter | Change end-of-statement delimiter. | +| /exit | /q | /exit | Exit. | +| /help | /? | /help [term] | Show this table, or search for help on a term. | +| /nopager | /n | /nopager | Disable pager; print to stdout. | +| /notee | | /notee | Stop writing results to an output file. | +| /nowarnings | /w | /nowarnings | Disable automatic warnings display. | +| /pager | /P | /pager [command] | Set pager to [command]. Print query results via pager. | +| /prompt | /R | /prompt | Change prompt format. | +| /quit | /q | /quit | Quit. | +| /redirectformat | /Tr | /redirectformat | Change the table format used to output redirected results. | +| /rehash | /# | /rehash | Refresh auto-completions. | +| /source | /. | /source | Execute queries from a file. | +| /status | /s | /status | Get status information from the server. | +| /system | | /system [-r] | Execute a system shell command (raw mode with -r). | +| /tableformat | /T | /tableformat | Change the table format used to output interactive results. | +| /tee | | /tee [-o] | Append all results to an output file (overwrite using -o). | +| /use | /u | /use | Change to a new database. | +| /warnings | /W | /warnings | Enable automatic warnings display. | +| /watch | | /watch [seconds] [-c] | Execute query every [seconds] seconds (5 by default). | ++-----------------+----------+-----------------------------------+-------------------------------------------------------------+ diff --git a/test/pytests/test_dsn_aliases.py b/test/pytests/test_dsn_aliases.py new file mode 100644 index 00000000..e593a059 --- /dev/null +++ b/test/pytests/test_dsn_aliases.py @@ -0,0 +1,102 @@ +from collections.abc import Mapping + +from mycli.packages.special.dsn_aliases import DsnAliases + + +class DummyConfig(dict): + def __init__(self, initial: Mapping[str, object] | None = None) -> None: + super().__init__(initial or {}) + self.encoding: str | None = None + self.write_calls = 0 + + def write(self) -> None: + self.write_calls += 1 + + +def test_from_config_returns_instance_with_same_config() -> None: + config = DummyConfig() + + aliases = DsnAliases.from_config(config) + + assert isinstance(aliases, DsnAliases) + assert aliases.config is config + + +def test_list_and_get_use_alias_dsn_section() -> None: + config = DummyConfig({ + 'alias_dsn': { + 'prod': 'mysql://prod/db', + 'staging': 'mysql://staging/db', + }, + }) + aliases = DsnAliases(config) + + assert aliases.list() == ['prod', 'staging'] + assert aliases.get('prod') == 'mysql://prod/db' + assert aliases.get('missing') is None + + +def test_list_returns_empty_list_when_section_is_missing() -> None: + aliases = DsnAliases(DummyConfig()) + + assert aliases.list() == [] + + +def test_save_creates_section_sets_encoding_and_writes_config() -> None: + config = DummyConfig() + aliases = DsnAliases(config) + + result = aliases.save('prod', 'mysql://prod/db') + + assert result == 'Saved: prod' + assert config.encoding == 'utf-8' + assert config == {'alias_dsn': {'prod': 'mysql://prod/db'}} + assert config.write_calls == 1 + + +def test_save_updates_existing_section_and_writes_config() -> None: + config = DummyConfig({'alias_dsn': {'prod': 'mysql://prod/db'}}) + aliases = DsnAliases(config) + + result = aliases.save('staging', 'mysql://staging/db') + + assert result == 'Saved: staging' + assert config.encoding == 'utf-8' + assert config['alias_dsn'] == { + 'prod': 'mysql://prod/db', + 'staging': 'mysql://staging/db', + } + assert config.write_calls == 1 + + +def test_delete_removes_existing_alias_and_writes_config() -> None: + config = DummyConfig({'alias_dsn': {'prod': 'mysql://prod/db'}}) + aliases = DsnAliases(config) + + result = aliases.delete('prod') + + assert result == 'Deleted: prod' + assert config['alias_dsn'] == {} + assert config.write_calls == 1 + + +def test_delete_returns_not_found_without_writing_config() -> None: + config = DummyConfig({'alias_dsn': {'prod': 'mysql://prod/db'}}) + aliases = DsnAliases(config) + + result = aliases.delete('missing') + + assert result == 'Not Found: missing' + assert config['alias_dsn'] == {'prod': 'mysql://prod/db'} + assert config.write_calls == 0 + + +def test_delete_returns_not_found_when_section_is_missing() -> None: + config = DummyConfig() + aliases = DsnAliases(config) + + result = aliases.delete('missing') + + assert result == 'Not Found: missing' + assert config == {} + assert config.write_calls == 0 diff --git a/test/pytests/test_smart_completion_public_schema_only.py b/test/pytests/test_smart_completion_public_schema_only.py index ab329c6c..5c4ab2df 100644 --- a/test/pytests/test_smart_completion_public_schema_only.py +++ b/test/pytests/test_smart_completion_public_schema_only.py @@ -99,7 +99,10 @@ def test_special_name_completion(completer, complete_event): text = "\\d" position = len("\\d") result = completer.get_completions(Document(text=text, cursor_position=position), complete_event) - assert list(result) == [Completion(text="\\dt", start_position=-2)] + assert list(result) == [ + Completion(text="\\dt", start_position=-2), + Completion(text="\\dsn", start_position=-2), + ] def test_empty_string_completion(completer, complete_event): diff --git a/test/pytests/test_special_iocommands.py b/test/pytests/test_special_iocommands.py index 00d14d27..b37e6fdd 100644 --- a/test/pytests/test_special_iocommands.py +++ b/test/pytests/test_special_iocommands.py @@ -44,6 +44,30 @@ def delete(self, name: str) -> str: return f'{name}: Deleted.' +class FakeDsnAliases: + usage = '\nFAKE DSN USAGE' + + def __init__(self, aliases: dict[str, str] | None = None) -> None: + self.aliases = {} if aliases is None else dict(aliases) + self.saved: list[tuple[str, str]] = [] + self.deleted: list[str] = [] + + def list(self) -> list[str]: + return list(self.aliases) + + def get(self, alias: str) -> str | None: + return self.aliases.get(alias) + + def save(self, alias: str, dsn: str) -> str: + self.saved.append((alias, dsn)) + self.aliases[alias] = dsn + return f'Saved: {alias}' + + def delete(self, alias: str) -> str: + self.deleted.append(alias) + return f'Deleted: {alias}' + + class FakeCursor: def __init__(self, descriptions: dict[str, list[tuple[str]] | None] | None = None) -> None: self.descriptions = {} if descriptions is None else descriptions @@ -108,6 +132,8 @@ def reset_iocommands_state(monkeypatch) -> Generator[None, None, None]: original_favoritequeries = iocommands.favoritequeries had_instance = hasattr(iocommands.FavoriteQueries, 'instance') original_instance = getattr(iocommands.FavoriteQueries, 'instance', None) + had_dsn_instance = hasattr(iocommands.DsnAliases, 'instance') + original_dsn_instance = getattr(iocommands.DsnAliases, 'instance', None) yield @@ -129,6 +155,8 @@ def reset_iocommands_state(monkeypatch) -> Generator[None, None, None]: iocommands.favoritequeries = original_favoritequeries if had_instance: iocommands.FavoriteQueries.instance = original_instance + if had_dsn_instance: + iocommands.DsnAliases.instance = original_dsn_instance @pytest.fixture @@ -681,6 +709,82 @@ def test_list_substitute_save_delete_and_redirect_state(tmp_path: Path, monkeypa assert iocommands.is_redirected() is True +def test_dsn_command_shows_current_connection(monkeypatch) -> None: + monkeypatch.setattr(iocommands, 'compute_current_dsn', lambda cur: 'mysql://user@host/db') + monkeypatch.setattr(iocommands.DsnAliases, 'instance', FakeDsnAliases(), raising=False) + + result = iocommands.dsn(cur=FakeCursor(), arg='show')[0] + + assert result.header == ['Current Connection'] + assert result.rows == [('mysql://user@host/db',)] + + +def test_dsn_command_lists_aliases(monkeypatch) -> None: + aliases = FakeDsnAliases({'prod': 'mysql://prod/db'}) + monkeypatch.setattr(iocommands.DsnAliases, 'instance', aliases, raising=False) + + result = iocommands.dsn(cur=FakeCursor(), arg='list')[0] + + assert result.status is None + assert result.header == ['Alias', 'DSN'] + assert result.rows == [('prod', 'mysql://prod/db')] + + +def test_dsn_command_reports_empty_alias_list(monkeypatch) -> None: + monkeypatch.setattr(iocommands.DsnAliases, 'instance', FakeDsnAliases(), raising=False) + + result = iocommands.dsn(cur=FakeCursor(), arg='list')[0] + + assert result.status == 'No DSN Aliases found.' + assert result.header == ['Alias', 'DSN'] + assert result.rows == [] + + +def test_dsn_command_saves_current_connection(monkeypatch) -> None: + aliases = FakeDsnAliases() + monkeypatch.setattr(iocommands, 'compute_current_dsn', lambda cur: 'mysql://user@host/db') + monkeypatch.setattr(iocommands.DsnAliases, 'instance', aliases, raising=False) + + result = iocommands.dsn(cur=FakeCursor(), arg='save prod')[0] + + assert result.status == 'Saved: prod' + assert aliases.saved == [('prod', 'mysql://user@host/db')] + + +def test_dsn_command_rejects_save_without_single_alias(monkeypatch) -> None: + monkeypatch.setattr(iocommands.DsnAliases, 'instance', FakeDsnAliases(), raising=False) + + assert iocommands.dsn(cur=FakeCursor(), arg='save')[0].status == 'Error: a single alias-name argument is required to save.' + assert iocommands.dsn(cur=FakeCursor(), arg='save one two')[0].status == ('Error: a single alias-name argument is required to save.') + + +def test_dsn_command_deletes_alias(monkeypatch) -> None: + aliases = FakeDsnAliases({'prod': 'mysql://prod/db'}) + monkeypatch.setattr(iocommands.DsnAliases, 'instance', aliases, raising=False) + + result = iocommands.dsn(cur=FakeCursor(), arg='delete prod')[0] + + assert result.status == 'Deleted: prod' + assert aliases.deleted == ['prod'] + + +def test_dsn_command_rejects_delete_without_single_alias(monkeypatch) -> None: + monkeypatch.setattr(iocommands.DsnAliases, 'instance', FakeDsnAliases(), raising=False) + + assert iocommands.dsn(cur=FakeCursor(), arg='delete')[0].status == ('Error: a single alias-name argument is required to delete.') + assert iocommands.dsn(cur=FakeCursor(), arg='delete one two')[0].status == ( + 'Error: a single alias-name argument is required to delete.' + ) + + +def test_dsn_command_shows_usage_for_help_and_unknown_subcommands(monkeypatch) -> None: + aliases = FakeDsnAliases() + monkeypatch.setattr(iocommands.DsnAliases, 'instance', aliases, raising=False) + + assert iocommands.dsn(cur=FakeCursor(), arg='help')[0].preamble == aliases.usage + assert iocommands.dsn(cur=FakeCursor(), arg='unknown')[0].preamble == aliases.usage + + def test_execute_system_command_usage_parse_and_cd(monkeypatch) -> None: usage = 'Syntax: system [-r] [command].\n-r denotes "raw" mode, in which output is passed through without formatting.' assert iocommands.execute_system_command('')[0].status == usage diff --git a/test/pytests/test_special_utils.py b/test/pytests/test_special_utils.py index a06e81b4..0ed21e13 100644 --- a/test/pytests/test_special_utils.py +++ b/test/pytests/test_special_utils.py @@ -3,6 +3,7 @@ import os import pathlib import tempfile +from types import SimpleNamespace from unittest.mock import MagicMock import pymysql @@ -11,6 +12,7 @@ import mycli.packages.special.utils from mycli.packages.special.utils import ( CACHED_SSL_VERSION, + compute_current_dsn, format_uptime, get_local_timezone, get_server_timezone, @@ -278,3 +280,45 @@ def astimezone(self) -> FakeAwareDatetime: monkeypatch.setattr(mycli.packages.special.utils.datetime, 'datetime', FakeDatetime) assert get_local_timezone() == '' + + +def test_compute_current_dsn_for_tcp_connection() -> None: + connection = SimpleNamespace( + user='user@example.com', + host='db.example.com', + port=3307, + db='my db', + unix_socket=None, + charset='utf8mb4', + ) + cursor = SimpleNamespace(connection=connection) + + assert compute_current_dsn(cursor) == 'mysql://user%40example.com@db.example.com:3307/my%20db' + + +def test_compute_current_dsn_for_socket_connection() -> None: + connection = SimpleNamespace( + user='alice', + host='localhost', + port=3306, + db='', + unix_socket='/tmp/mysql.sock', + charset='utf8mb4', + ) + cursor = SimpleNamespace(connection=connection) + + assert compute_current_dsn(cursor) == 'mysql://alice@localhost?socket=%2Ftmp%2Fmysql.sock' + + +def test_compute_current_dsn_includes_non_default_character_set() -> None: + connection = SimpleNamespace( + user='alice', + host=None, + port=3306, + db=b'mysql', + unix_socket=None, + charset='latin1', + ) + cursor = SimpleNamespace(connection=connection) + + assert compute_current_dsn(cursor) == 'mysql://alice@localhost:3306/mysql?character_set=latin1'