From e773f0f23a89a89418a21ee1cc8a0be9127ab98c Mon Sep 17 00:00:00 2001 From: Roland Walker Date: Tue, 4 Aug 2026 05:02:24 -0400 Subject: [PATCH] don't rewrite myclirc comments on /dsn save The user's own commentary could be rewritten from the package defaults if saving or deleting a named DSN via the REPL. --- changelog.md | 1 + mycli/client.py | 2 +- mycli/packages/special/dsn_aliases.py | 65 +++++++-- test/pytests/test_client.py | 12 ++ test/pytests/test_dsn_aliases.py | 194 +++++++++++++++++++++++++- 5 files changed, 261 insertions(+), 13 deletions(-) diff --git a/changelog.md b/changelog.md index 31a1757a..5a7777d0 100644 --- a/changelog.md +++ b/changelog.md @@ -9,6 +9,7 @@ Features Bugfixes --------- * Don't allow saving favorite queries to rewrite comments in `~/.myclirc`. +* Don't allow saving named DSNs to rewrite comments in `~/.myclirc`. Documentation diff --git a/mycli/client.py b/mycli/client.py index 722639a7..287646d9 100644 --- a/mycli/client.py +++ b/mycli/client.py @@ -116,7 +116,7 @@ def __init__( self.default_keepalive_ticks = c['connection'].as_int('default_keepalive_ticks') FavoriteQueries.instance = FavoriteQueries.from_config(self.config, myclirc) - DsnAliases.instance = DsnAliases.from_config(self.config, self) + DsnAliases.instance = DsnAliases.from_config(self.config, self, config_file=myclirc) 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 index 3f704274..8086d1ff 100644 --- a/mycli/packages/special/dsn_aliases.py +++ b/mycli/packages/special/dsn_aliases.py @@ -1,9 +1,10 @@ from __future__ import annotations +import os from typing import TYPE_CHECKING, Any from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit -from mycli.config import str_to_bool +from mycli.config import read_config_file, str_to_bool from mycli.constants import DEFAULT_CHARSET, DEFAULT_PROMPT, KNOWN_DSN_QUERY_PARAMS if TYPE_CHECKING: @@ -11,6 +12,7 @@ DSN_SUBCOMMANDS = {'help', 'list', 'show', 'save', 'delete'} INVALID_DSN_ALIAS_ERROR = 'Error: DSN aliases cannot start with a dash.' +MISSING = object() SSL_QUERY_PARAMS = { 'ssl_ca': 'ca', @@ -86,13 +88,28 @@ class DsnAliases: # Class-level variable, for convenience to use as a singleton. instance: DsnAliases - def __init__(self, config: Any, mycli: MyCli | None = None) -> None: + def __init__(self, config: Any, mycli: MyCli | None = None, config_file: str | None = None) -> None: self.config = config self.mycli = mycli + self.config_file = config_file @classmethod - def from_config(cls, config: Any, mycli: MyCli | None = None) -> DsnAliases: - return DsnAliases(config, mycli) + def from_config(cls, config: Any, mycli: MyCli | None = None, config_file: str | None = None) -> DsnAliases: + return DsnAliases(config, mycli, config_file) + + def _config_for_write(self) -> Any: + if self.config_file is None: + return self.config + + config = read_config_file(self.config_file) + if config is None: + raise OSError(f"Unable to read config file '{os.path.expanduser(self.config_file)}'.") + return config + + def _set_alias(self, config: Any, alias: str, dsn: str) -> None: + if self.section_name not in config: + config[self.section_name] = {} + config[self.section_name][alias] = dsn def _query_param_defaults(self) -> dict[str, Any]: if self.mycli is None: @@ -155,19 +172,45 @@ def get(self, alias: str) -> str | None: def save(self, alias: str, dsn: str) -> str: if not is_valid_dsn_alias(alias): return INVALID_DSN_ALIAS_ERROR - 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() + + config = self._config_for_write() + config.encoding = 'utf-8' + section_existed = self.section_name in config + previous_dsn = config.get(self.section_name, {}).get(alias, MISSING) + self._set_alias(config, alias, dsn) + try: + config.write() + except Exception: + if previous_dsn is MISSING: + del config[self.section_name][alias] + if not section_existed: + del config[self.section_name] + else: + config[self.section_name][alias] = previous_dsn + raise + + if config is not self.config: + self._set_alias(self.config, alias, dsn) return f'Saved: {alias}' def delete(self, alias: str) -> str: if not is_valid_dsn_alias(alias): return INVALID_DSN_ALIAS_ERROR try: - del self.config[self.section_name][alias] + self.config[self.section_name][alias] except KeyError: return f'Not Found: {alias}' - self.config.write() + + config = self._config_for_write() + if alias in config.get(self.section_name, {}): + dsn = config[self.section_name][alias] + del config[self.section_name][alias] + try: + config.write() + except Exception: + config[self.section_name][alias] = dsn + raise + + if config is not self.config: + del self.config[self.section_name][alias] return f'Deleted: {alias}' diff --git a/test/pytests/test_client.py b/test/pytests/test_client.py index 752521de..9c3d1b3d 100644 --- a/test/pytests/test_client.py +++ b/test/pytests/test_client.py @@ -9,6 +9,7 @@ import mycli.client as client_module from mycli.client import MyCli +from mycli.packages.special.dsn_aliases import DsnAliases from mycli.packages.special.favoritequeries import FavoriteQueries @@ -185,6 +186,17 @@ def test_init_configures_favorite_queries_with_user_config_path(monkeypatch: pyt assert FavoriteQueries.instance.config_file == myclirc +def test_init_configures_dsn_aliases_with_user_config_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + patch_constructor_side_effects(monkeypatch) + myclirc = write_myclirc(tmp_path, '') + + cli = MyCli(myclirc=myclirc) + + assert DsnAliases.instance.config is cli.config + assert DsnAliases.instance.mycli is cli + assert DsnAliases.instance.config_file == myclirc + + def test_init_uses_default_myclirc_when_xdg_config_is_missing(monkeypatch: pytest.MonkeyPatch) -> None: patch_constructor_side_effects(monkeypatch) config_file_args: list[list[str | Any]] = [] diff --git a/test/pytests/test_dsn_aliases.py b/test/pytests/test_dsn_aliases.py index a01f5170..b2eee41f 100644 --- a/test/pytests/test_dsn_aliases.py +++ b/test/pytests/test_dsn_aliases.py @@ -1,9 +1,13 @@ from collections.abc import Mapping +from pathlib import Path from types import SimpleNamespace from typing import Any from urllib.parse import parse_qsl, urlsplit +import pytest + from mycli.constants import KNOWN_DSN_QUERY_PARAMS +import mycli.packages.special.dsn_aliases as dsn_aliases_module from mycli.packages.special.dsn_aliases import INVALID_DSN_ALIAS_ERROR, DsnAliases, is_valid_dsn_alias @@ -17,6 +21,11 @@ def write(self) -> None: self.write_calls += 1 +class FailingConfig(DummyConfig): + def write(self) -> None: + raise OSError('write failed') + + def test_is_valid_dsn_alias_rejects_dash_prefix() -> None: assert is_valid_dsn_alias('prod') is True assert is_valid_dsn_alias('-prod') is False @@ -25,10 +34,11 @@ def test_is_valid_dsn_alias_rejects_dash_prefix() -> None: def test_from_config_returns_instance_with_same_config() -> None: config = DummyConfig() - aliases = DsnAliases.from_config(config) + aliases = DsnAliases.from_config(config, config_file='/tmp/myclirc') assert isinstance(aliases, DsnAliases) assert aliases.config is config + assert aliases.config_file == '/tmp/myclirc' def test_from_config_retains_mycli_runtime() -> None: @@ -161,6 +171,188 @@ def test_delete_returns_not_found_when_section_is_missing() -> None: assert config.write_calls == 0 +def test_save_preserves_user_config_comments_and_excludes_merged_values(tmp_path: Path) -> None: + config_file = tmp_path / 'myclirc' + config_file.write_text( + """# User introduction. +[main] +prompt = custom # Inline comment. + +[alias_dsn] +# Existing alias. +existing = mysql://existing/db +# User footer. +""", + encoding='utf-8', + ) + merged_config = DummyConfig({ + 'main': {'prompt': 'custom', 'package_default': 'do not write'}, + 'alias_dsn': {'existing': 'mysql://existing/db'}, + }) + aliases = DsnAliases(merged_config, config_file=str(config_file)) + + result = aliases.save('new', 'mysql://new/db') + + assert result == 'Saved: new' + assert ( + config_file.read_text(encoding='utf-8') + == """# User introduction. +[main] +prompt = custom# Inline comment. + +[alias_dsn] +# Existing alias. +existing = mysql://existing/db +new = mysql://new/db +# User footer. +""" + ) + assert merged_config['alias_dsn']['new'] == 'mysql://new/db' + assert 'package_default' not in config_file.read_text(encoding='utf-8') + + +def test_save_reloads_user_config_before_writing(tmp_path: Path) -> None: + config_file = tmp_path / 'myclirc' + config_file.write_text('[alias_dsn]\nexisting = mysql://existing/db\n', encoding='utf-8') + merged_config = DummyConfig({'alias_dsn': {'existing': 'mysql://existing/db'}}) + aliases = DsnAliases(merged_config, config_file=str(config_file)) + config_file.write_text( + '# Added while mycli is running.\n[alias_dsn]\nexisting = mysql://existing/db\nexternal = mysql://external/db\n', + encoding='utf-8', + ) + + aliases.save('new', 'mysql://new/db') + + contents = config_file.read_text(encoding='utf-8') + assert contents.startswith('# Added while mycli is running.\n') + assert 'external = mysql://external/db\n' in contents + assert 'new = mysql://new/db\n' in contents + + +def test_save_overwrites_alias_without_removing_its_comment(tmp_path: Path) -> None: + config_file = tmp_path / 'myclirc' + config_file.write_text( + """[alias_dsn] +# Keep this explanation. +prod = mysql://old/db +""", + encoding='utf-8', + ) + merged_config = DummyConfig({'alias_dsn': {'prod': 'mysql://old/db'}}) + + DsnAliases(merged_config, config_file=str(config_file)).save('prod', 'mysql://new/db') + + assert ( + config_file.read_text(encoding='utf-8') + == """[alias_dsn] +# Keep this explanation. +prod = mysql://new/db +""" + ) + assert merged_config['alias_dsn']['prod'] == 'mysql://new/db' + + +def test_delete_preserves_unrelated_user_config_comments(tmp_path: Path) -> None: + config_file = tmp_path / 'myclirc' + config_file.write_text( + """# User introduction. +[alias_dsn] +# Removed with the alias. +remove = mysql://remove/db +# Keep this explanation. +keep = mysql://keep/db +# User footer. +""", + encoding='utf-8', + ) + merged_config = DummyConfig({'alias_dsn': {'remove': 'mysql://remove/db', 'keep': 'mysql://keep/db'}}) + aliases = DsnAliases(merged_config, config_file=str(config_file)) + + result = aliases.delete('remove') + + assert result == 'Deleted: remove' + assert ( + config_file.read_text(encoding='utf-8') + == """# User introduction. +[alias_dsn] +# Keep this explanation. +keep = mysql://keep/db +# User footer. +""" + ) + assert merged_config['alias_dsn'] == {'keep': 'mysql://keep/db'} + + +def test_delete_effective_system_alias_does_not_rewrite_user_config(tmp_path: Path) -> None: + config_file = tmp_path / 'myclirc' + original = '# User commentary.\n[main]\nprompt = custom\n' + config_file.write_text(original, encoding='utf-8') + merged_config = DummyConfig({'alias_dsn': {'system': 'mysql://system/db'}}) + aliases = DsnAliases(merged_config, config_file=str(config_file)) + + result = aliases.delete('system') + + assert result == 'Deleted: system' + assert config_file.read_text(encoding='utf-8') == original + assert merged_config['alias_dsn'] == {} + + +def test_invalid_alias_does_not_read_user_config(monkeypatch: pytest.MonkeyPatch) -> None: + aliases = DsnAliases(DummyConfig(), config_file='~/.myclirc') + monkeypatch.setattr( + dsn_aliases_module, + 'read_config_file', + lambda _path: pytest.fail('invalid aliases must not read the user config'), + ) + + assert aliases.save('-prod', 'mysql://prod/db') == INVALID_DSN_ALIAS_ERROR + assert aliases.delete('-prod') == INVALID_DSN_ALIAS_ERROR + + +def test_save_does_not_update_runtime_config_when_user_config_cannot_be_read( + monkeypatch: pytest.MonkeyPatch, +) -> None: + merged_config = DummyConfig({'alias_dsn': {'existing': 'mysql://existing/db'}}) + aliases = DsnAliases(merged_config, config_file='~/.myclirc') + monkeypatch.setattr(dsn_aliases_module, 'read_config_file', lambda _path: None) + + with pytest.raises(OSError, match=r"Unable to read config file '.*/\.myclirc'\."): + aliases.save('new', 'mysql://new/db') + + assert merged_config['alias_dsn'] == {'existing': 'mysql://existing/db'} + + +@pytest.mark.parametrize('initial', ({}, {'alias_dsn': {'existing': 'mysql://existing/db'}})) +def test_save_restores_runtime_config_after_write_failure(initial: dict[str, object]) -> None: + config = FailingConfig(initial) + aliases = DsnAliases(config) + + with pytest.raises(OSError, match='write failed'): + aliases.save('new', 'mysql://new/db') + + assert config == initial + + +def test_save_restores_overwritten_runtime_alias_after_write_failure() -> None: + config = FailingConfig({'alias_dsn': {'existing': 'mysql://existing/db'}}) + aliases = DsnAliases(config) + + with pytest.raises(OSError, match='write failed'): + aliases.save('existing', 'mysql://new/db') + + assert config['alias_dsn']['existing'] == 'mysql://existing/db' + + +def test_delete_restores_runtime_alias_after_write_failure() -> None: + config = FailingConfig({'alias_dsn': {'existing': 'mysql://existing/db'}}) + aliases = DsnAliases(config) + + with pytest.raises(OSError, match='write failed'): + aliases.delete('existing') + + assert config['alias_dsn'] == {'existing': 'mysql://existing/db'} + + def test_dsn_more_adds_non_default_runtime_parameters_in_sorted_order() -> None: config = DummyConfig({ 'main': {'prompt': 'configured> '},