Don't rewrite ~/.myclirc comments on /dsn save or /dsn delete

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

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"])

mycli/packages/special/dsn_aliases.py

@@ -1,16 +1,18 @@
 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:
     from mycli.client import MyCli
 
 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}'

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]] = []

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> '},