Move --list-ssh-config out of main.py

changelog.md

@@ -38,6 +38,7 @@ Internal
 * Move `--checkup` logic to the new `main_modes` with `--batch`.
 * Move `--execute` logic to the new `main_modes` with `--batch`.
 * Move `--list-dsn` logic to the new `main_modes` with `--batch`.
+* Move `--list-ssh-config` logic to the new `main_modes` with `--batch`.
 * Sort coverage report in tox suite.
 * Skip more tests when a database connection is not present.
 

mycli/main.py

@@ -88,6 +88,7 @@
 from mycli.main_modes.checkup import main_checkup
 from mycli.main_modes.execute import main_execute_from_cli
 from mycli.main_modes.list_dsn import main_list_dsn
+from mycli.main_modes.list_ssh_config import main_list_ssh_config
 from mycli.packages import special
 from mycli.packages.filepaths import dir_path_exists, guess_socket_location
 from mycli.packages.hybrid_redirection import get_redirect_components, is_redirect_command
@@ -98,16 +99,12 @@
 from mycli.packages.special.main import ArgType
 from mycli.packages.special.utils import format_uptime, get_ssl_version, get_uptime, get_warning_count
 from mycli.packages.sqlresult import SQLResult
+from mycli.packages.ssh_utils import read_ssh_config
 from mycli.packages.string_utils import sanitize_terminal_title
 from mycli.packages.tabular_output import sql_format
 from mycli.sqlcompleter import SQLCompleter
 from mycli.sqlexecute import FIELD_TYPES, SQLExecute
 
-try:
-    import paramiko
-except ImportError:
-    from mycli.packages.paramiko_stub import paramiko  # type: ignore[no-redef]
-
 sqlparse.engine.grouping.MAX_GROUPING_DEPTH = None  # type: ignore[assignment]
 sqlparse.engine.grouping.MAX_GROUPING_TOKENS = None  # type: ignore[assignment]
 
@@ -2316,19 +2313,7 @@ def get_password_from_file(password_file: str | None) -> str | None:
         sys.exit(main_list_dsn(mycli, cli_args))
 
     if cli_args.list_ssh_config:
-        ssh_config = read_ssh_config(cli_args.ssh_config_path)
-        try:
-            host_entries = ssh_config.get_hostnames()
-        except KeyError:
-            click.secho('Error reading ssh config', err=True, fg="red")
-            sys.exit(1)
-        for host_entry in host_entries:
-            if cli_args.verbose:
-                host_config = ssh_config.lookup(host_entry)
-                click.secho(f"{host_entry} : {host_config.get('hostname')}")
-            else:
-                click.secho(host_entry)
-        sys.exit(0)
+        sys.exit(main_list_ssh_config(mycli, cli_args))
 
     if 'MYSQL_UNIX_PORT' in os.environ:
         # deprecated 2026-03
@@ -2761,24 +2746,6 @@ def edit_and_execute(event: KeyPressEvent) -> None:
     buff.open_in_editor(validate_and_handle=False)
 
 
-def read_ssh_config(ssh_config_path: str):
-    ssh_config = paramiko.config.SSHConfig()
-    try:
-        with open(ssh_config_path) as f:
-            ssh_config.parse(f)
-    except FileNotFoundError as e:
-        click.secho(str(e), err=True, fg="red")
-        sys.exit(1)
-    # Paramiko prior to version 2.7 raises Exception on parse errors.
-    # In 2.7 it has become paramiko.ssh_exception.SSHException,
-    # but let's catch everything for compatibility
-    except Exception as err:
-        click.secho(f"Could not parse SSH configuration file {ssh_config_path}:\n{err} ", err=True, fg="red")
-        sys.exit(1)
-    else:
-        return ssh_config
-
-
 def filtered_sys_argv() -> list[str]:
     args = sys.argv[1:]
     if args == ['-h']:

mycli/main_modes/list_ssh_config.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+from mycli.packages.ssh_utils import read_ssh_config
+
+if TYPE_CHECKING:
+    from mycli.main import CliArgs, MyCli
+
+
+def main_list_ssh_config(mycli: 'MyCli', cli_args: 'CliArgs') -> int:
+    ssh_config = read_ssh_config(cli_args.ssh_config_path)
+    try:
+        host_entries = ssh_config.get_hostnames()
+    except KeyError:
+        click.secho('Error reading ssh config', err=True, fg="red")
+        return 1
+    for host_entry in host_entries:
+        if cli_args.verbose:
+            host_config = ssh_config.lookup(host_entry)
+            click.secho(f"{host_entry} : {host_config.get('hostname')}")
+        else:
+            click.secho(host_entry)
+    return 0

mycli/packages/ssh_utils.py

@@ -0,0 +1,27 @@
+import sys
+
+import click
+
+try:
+    import paramiko
+except ImportError:
+    from mycli.packages.paramiko_stub import paramiko  # type: ignore[no-redef]
+
+
+# it isn't cool that this utility function can exit(), but it is slated to be removed anyway
+def read_ssh_config(ssh_config_path: str):
+    ssh_config = paramiko.config.SSHConfig()
+    try:
+        with open(ssh_config_path) as f:
+            ssh_config.parse(f)
+    except FileNotFoundError as e:
+        click.secho(str(e), err=True, fg="red")
+        sys.exit(1)
+    # Paramiko prior to version 2.7 raises Exception on parse errors.
+    # In 2.7 it has become paramiko.ssh_exception.SSHException,
+    # but let's catch everything for compatibility
+    except Exception as err:
+        click.secho(f"Could not parse SSH configuration file {ssh_config_path}:\n{err} ", err=True, fg="red")
+        sys.exit(1)
+    else:
+        return ssh_config

test/pytests/test_main_modes_list_ssh_config.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, cast
+
+import mycli.main_modes.list_ssh_config as list_ssh_config_mode
+
+
+@dataclass
+class DummyCliArgs:
+    ssh_config_path: str = 'ssh_config'
+    verbose: bool = False
+
+
+class DummySSHConfig:
+    def __init__(self, hostnames: list[str] | Exception, lookups: dict[str, dict[str, str]] | None = None) -> None:
+        self.hostnames = hostnames
+        self.lookups = lookups or {}
+
+    def get_hostnames(self) -> list[str]:
+        if isinstance(self.hostnames, Exception):
+            raise self.hostnames
+        return self.hostnames
+
+    def lookup(self, hostname: str) -> dict[str, str]:
+        return self.lookups[hostname]
+
+
+def main_list_ssh_config(cli_args: DummyCliArgs) -> int:
+    return list_ssh_config_mode.main_list_ssh_config(cast(Any, object()), cast(Any, cli_args))
+
+
+def test_main_list_ssh_config_lists_hostnames(monkeypatch) -> None:
+    secho_calls: list[tuple[str, bool | None, str | None]] = []
+    ssh_config = DummySSHConfig(['prod', 'staging'])
+
+    monkeypatch.setattr(list_ssh_config_mode, 'read_ssh_config', lambda _path: ssh_config)
+    monkeypatch.setattr(
+        list_ssh_config_mode.click,
+        'secho',
+        lambda message, err=None, fg=None: secho_calls.append((message, err, fg)),
+    )
+
+    result = main_list_ssh_config(DummyCliArgs(verbose=False))
+
+    assert result == 0
+    assert secho_calls == [
+        ('prod', None, None),
+        ('staging', None, None),
+    ]
+
+
+def test_main_list_ssh_config_lists_verbose_host_details(monkeypatch) -> None:
+    secho_calls: list[tuple[str, bool | None, str | None]] = []
+    ssh_config = DummySSHConfig(
+        ['prod'],
+        lookups={'prod': {'hostname': 'db.example.com'}},
+    )
+
+    monkeypatch.setattr(list_ssh_config_mode, 'read_ssh_config', lambda _path: ssh_config)
+    monkeypatch.setattr(
+        list_ssh_config_mode.click,
+        'secho',
+        lambda message, err=None, fg=None: secho_calls.append((message, err, fg)),
+    )
+
+    result = main_list_ssh_config(DummyCliArgs(verbose=True))
+
+    assert result == 0
+    assert secho_calls == [('prod : db.example.com', None, None)]
+
+
+def test_main_list_ssh_config_reports_host_lookup_errors(monkeypatch) -> None:
+    secho_calls: list[tuple[str, bool | None, str | None]] = []
+    ssh_config = DummySSHConfig(KeyError('bad ssh config'))
+
+    monkeypatch.setattr(list_ssh_config_mode, 'read_ssh_config', lambda _path: ssh_config)
+    monkeypatch.setattr(
+        list_ssh_config_mode.click,
+        'secho',
+        lambda message, err=None, fg=None: secho_calls.append((message, err, fg)),
+    )
+
+    result = main_list_ssh_config(DummyCliArgs())
+
+    assert result == 1
+    assert secho_calls == [('Error reading ssh config', True, 'red')]