diff --git a/AGENTS.md b/AGENTS.md index 92df4a43..52ebdbea 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,7 +39,6 @@ A command line client for MySQL with auto-completion and syntax highlighting. ├── mycli/packages/hybrid_redirection.py # implementation of shell-style redirects ├── mycli/packages/interactive_utils.py # utilities for confirming on destructive statements ├── mycli/packages/key_binding_utils.py # handlers for key bindings and related special commands -├── mycli/packages/paramiko_stub/ # stub in case the Paramiko library is not installed ├── mycli/packages/sql_utils.py # utilities for parsing SQL statements ├── mycli/packages/ptoolkit/ # extends prompt_toolkit ├── mycli/packages/special/ # implementation of mycli special commands diff --git a/changelog.md b/changelog.md index 69aee6fc..54c683de 100644 --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,11 @@ +Upcoming (TBD) +============== + +Breaking Changes +-------- +* Remove support for deprecated SSH jump functionality. + + 1.75.0 (2026/06/20) ============== diff --git a/mycli/cli_runner.py b/mycli/cli_runner.py index 6b35e657..85c7ead5 100644 --- a/mycli/cli_runner.py +++ b/mycli/cli_runner.py @@ -14,9 +14,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.cli_utils import is_valid_connection_scheme -from mycli.packages.ssh_utils import read_ssh_config if TYPE_CHECKING: from mycli.main import CliArgs @@ -73,30 +71,9 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non fg="yellow", ) - # ssh_port and ssh_config_path have truthy defaults and are not included - if ( - any([ - cli_args.ssh_user, - cli_args.ssh_host, - cli_args.ssh_password, - cli_args.ssh_key_filename, - cli_args.list_ssh_config, - cli_args.ssh_config_host, - ]) - and not cli_args.ssh_warning_off - ): - click.secho( - f"Warning: The built-in SSH functionality is deprecated and will be removed in a future release. See issue {ISSUES_URL}/1464", - err=True, - fg="red", - ) - if cli_args.list_dsn: sys.exit(main_list_dsn(mycli)) - if cli_args.list_ssh_config: - sys.exit(main_list_ssh_config(mycli, cli_args)) - if 'MYSQL_UNIX_PORT' in os.environ: # deprecated 2026-03 click.secho( @@ -161,6 +138,11 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non mycli.dsn_alias = cli_args.dsn if dsn_uri: + is_valid_scheme, scheme = is_valid_connection_scheme(dsn_uri) + if not is_valid_scheme: + click.secho(f'Error: Unknown connection scheme provided for DSN URI ({scheme}://)', err=True, fg='red') + sys.exit(1) + uri = urlparse(dsn_uri) if not database: database = uri.path[1:] # ignore the leading fwd slash @@ -255,23 +237,6 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non else: ssl = None - if cli_args.ssh_config_host: - ssh_config = read_ssh_config(cli_args.ssh_config_path).lookup(cli_args.ssh_config_host) - ssh_host = cli_args.ssh_host if cli_args.ssh_host else ssh_config.get("hostname") - ssh_user = cli_args.ssh_user if cli_args.ssh_user else ssh_config.get("user") - if ssh_config.get("port") and cli_args.ssh_port == 22: - # port has a default value, overwrite it if it's in the config - ssh_port = int(ssh_config.get("port")) - else: - ssh_port = cli_args.ssh_port - ssh_key_filename = cli_args.ssh_key_filename if cli_args.ssh_key_filename else ssh_config.get("identityfile", [None])[0] - else: - ssh_host = cli_args.ssh_host - ssh_user = cli_args.ssh_user - ssh_port = cli_args.ssh_port - ssh_key_filename = cli_args.ssh_key_filename - - ssh_key_filename = ssh_key_filename and os.path.expanduser(ssh_key_filename) # Merge init-commands: global, DSN-specific, then CLI init_cmds: list[str] = [] # 1) Global init-commands @@ -391,11 +356,6 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non socket=cli_args.socket, local_infile=cli_args.local_infile, ssl=ssl, - ssh_user=ssh_user, - ssh_host=ssh_host, - ssh_port=ssh_port, - ssh_password=cli_args.ssh_password, - ssh_key_filename=ssh_key_filename, init_command=combined_init_cmd, unbuffered=cli_args.unbuffered, character_set=cli_args.character_set, diff --git a/mycli/client_connection.py b/mycli/client_connection.py index 2eba7b05..b987c018 100644 --- a/mycli/client_connection.py +++ b/mycli/client_connection.py @@ -54,11 +54,6 @@ def connect( character_set: str | None = "", local_infile: bool | None = False, ssl: dict[str, Any] | None = None, - ssh_user: str | None = "", - ssh_host: str | None = "", - ssh_port: int = 22, - ssh_password: str | None = "", - ssh_key_filename: str | None = "", init_command: str | None = "", unbuffered: bool | None = None, use_keyring: bool | None = None, @@ -200,11 +195,6 @@ def connect( "character_set": character_set, "local_infile": use_local_infile, "ssl": ssl_config_or_none, - "ssh_user": ssh_user, - "ssh_host": ssh_host, - "ssh_port": int(ssh_port) if ssh_port else None, - "ssh_password": ssh_password, - "ssh_key_filename": ssh_key_filename, "init_command": init_command, "unbuffered": unbuffered, } diff --git a/mycli/completion_refresher.py b/mycli/completion_refresher.py index 81e74060..cba79bcc 100644 --- a/mycli/completion_refresher.py +++ b/mycli/completion_refresher.py @@ -71,11 +71,6 @@ def _bg_refresh( e.character_set, e.local_infile, e.ssl, - e.ssh_user, - e.ssh_host, - e.ssh_port, - e.ssh_password, - e.ssh_key_filename, ) except pymysql.err.OperationalError: return diff --git a/mycli/main.py b/mycli/main.py index 6bd1b253..7ca6448c 100755 --- a/mycli/main.py +++ b/mycli/main.py @@ -86,44 +86,6 @@ class CliArgs: type=click.Path(), help='File or FIFO path containing the password to connect to the db if not specified otherwise.', ) - ssh_user: str | None = clickdc.option( - type=str, - help='User name to connect to ssh server.', - ) - ssh_host: str | None = clickdc.option( - type=str, - help='Host name to connect to ssh server.', - ) - ssh_port: int = clickdc.option( - type=int, - default=22, - help='Port to connect to ssh server.', - ) - ssh_password: str | None = clickdc.option( - type=str, - help='Password to connect to ssh server.', - ) - ssh_key_filename: str | None = clickdc.option( - type=str, - help='Private key filename (identify file) for the ssh connection.', - ) - ssh_config_path: str = clickdc.option( - type=str, - help='Path to ssh configuration.', - default=os.path.expanduser('~') + '/.ssh/config', - ) - ssh_config_host: str | None = clickdc.option( - type=str, - help='Host to connect to ssh server reading from ssh configuration.', - ) - list_ssh_config: bool = clickdc.option( - is_flag=True, - help='list ssh configurations in the ssh config (requires paramiko).', - ) - ssh_warning_off: bool = clickdc.option( - is_flag=True, - help='Suppress the SSH deprecation notice.', - ) ssl_mode: str = clickdc.option( type=click.Choice(['auto', 'on', 'off']), help='Set desired SSL behavior. auto=preferred if TCP/IP, on=required, off=off.', diff --git a/mycli/main_modes/list_ssh_config.py b/mycli/main_modes/list_ssh_config.py deleted file mode 100644 index 6927580b..00000000 --- a/mycli/main_modes/list_ssh_config.py +++ /dev/null @@ -1,27 +0,0 @@ -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.client import MyCli - from mycli.main import CliArgs - - -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 mycli.verbosity >= 1: - host_config = ssh_config.lookup(host_entry) - click.secho(f"{host_entry} : {host_config.get('hostname')}") - else: - click.secho(host_entry) - return 0 diff --git a/mycli/packages/cli_utils.py b/mycli/packages/cli_utils.py index 65950130..d78b9f37 100644 --- a/mycli/packages/cli_utils.py +++ b/mycli/packages/cli_utils.py @@ -15,7 +15,7 @@ def is_valid_connection_scheme(text: str) -> tuple[bool, str | None]: if "://" not in text: return False, None scheme = text.split("://")[0] - if scheme not in ("mysql", "mysqlx", "tcp", "socket", "ssh"): + if scheme not in ("mysql", "mysqlx", "tcp", "socket"): return False, scheme else: return True, None diff --git a/mycli/packages/paramiko_stub/__init__.py b/mycli/packages/paramiko_stub/__init__.py deleted file mode 100644 index da2eca04..00000000 --- a/mycli/packages/paramiko_stub/__init__.py +++ /dev/null @@ -1,36 +0,0 @@ -"""A module to import instead of paramiko when it is not available (to avoid -checking for paramiko all over the place). - -When paramiko is first invoked, this simply shuts down mycli, telling the -user they either have to install paramiko or should not use SSH features. - -""" - - -class Paramiko: - def __getattr__(self, name: str) -> None: - import sys - from textwrap import dedent - - print( - dedent(""" - To enable certain SSH features you need to install ssh extras: - - pip install 'mycli[ssh]' - - or - - pip install paramiko sshtunnel - - This is required for the following command-line arguments: - - --list-ssh-config - --ssh-config-host - --ssh-host - """), - file=sys.stderr, - ) - sys.exit(1) - - -paramiko = Paramiko() diff --git a/mycli/packages/ssh_utils.py b/mycli/packages/ssh_utils.py deleted file mode 100644 index 1b81384a..00000000 --- a/mycli/packages/ssh_utils.py +++ /dev/null @@ -1,27 +0,0 @@ -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 diff --git a/mycli/schema_prefetcher.py b/mycli/schema_prefetcher.py index ac740ee9..7b7e4c77 100644 --- a/mycli/schema_prefetcher.py +++ b/mycli/schema_prefetcher.py @@ -224,11 +224,6 @@ def _make_executor(self) -> SQLExecute: sqlexecute.character_set, sqlexecute.local_infile, sqlexecute.ssl, - sqlexecute.ssh_user, - sqlexecute.ssh_host, - sqlexecute.ssh_port, - sqlexecute.ssh_password, - sqlexecute.ssh_key_filename, ) def _invalidate_app(self) -> None: diff --git a/mycli/sqlexecute.py b/mycli/sqlexecute.py index c4661b90..6b4f0049 100644 --- a/mycli/sqlexecute.py +++ b/mycli/sqlexecute.py @@ -19,12 +19,6 @@ from mycli.packages.special.main import CommandNotFound, execute from mycli.packages.sqlresult import SQLResult -try: - import paramiko # noqa: F401 - import sshtunnel -except ImportError: - pass - _logger = logging.getLogger(__name__) FIELD_TYPES = decoders.copy() @@ -171,11 +165,6 @@ def __init__( character_set: str | None, local_infile: bool | None, ssl: dict[str, Any] | None, - ssh_user: str | None, - ssh_host: str | None, - ssh_port: int | None, - ssh_password: str | None, - ssh_key_filename: str | None, init_command: str | None = None, unbuffered: bool | None = None, ) -> None: @@ -190,11 +179,6 @@ def __init__( self.ssl = ssl self.server_info: ServerInfo | None = None self.connection_id: int | None = None - self.ssh_user = ssh_user - self.ssh_host = ssh_host - self.ssh_port = ssh_port - self.ssh_password = ssh_password - self.ssh_key_filename = ssh_key_filename self.init_command = init_command self.unbuffered = unbuffered self.conn: Connection | None = None @@ -211,11 +195,6 @@ def connect( character_set: str | None = None, local_infile: bool | None = None, ssl: dict[str, Any] | None = None, - ssh_host: str | None = None, - ssh_port: int | None = None, - ssh_user: str | None = None, - ssh_password: str | None = None, - ssh_key_filename: str | None = None, init_command: str | None = None, unbuffered: bool | None = None, ): @@ -228,11 +207,6 @@ def connect( character_set = character_set if character_set is not None else self.character_set local_infile = local_infile if local_infile is not None else self.local_infile ssl = ssl if ssl is not None else self.ssl - ssh_user = ssh_user if ssh_user is not None else self.ssh_user - ssh_host = ssh_host if ssh_host is not None else self.ssh_host - ssh_port = ssh_port if ssh_port is not None else self.ssh_port - ssh_password = ssh_password if ssh_password is not None else self.ssh_password - ssh_key_filename = ssh_key_filename if ssh_key_filename is not None else self.ssh_key_filename init_command = init_command if init_command is not None else self.init_command unbuffered = unbuffered if unbuffered is not None else self.unbuffered _logger.debug( @@ -245,11 +219,6 @@ def connect( "\tcharacter_set: %r" "\tlocal_infile: %r" "\tssl: %r" - "\tssh_user: %r" - "\tssh_host: %r" - "\tssh_port: %r" - "\tssh_password: ***" - "\tssh_key_filename: %r" "\tinit_command: %r" "\tunbuffered: %r", db, @@ -260,10 +229,6 @@ def connect( character_set, local_infile, ssl, - ssh_user, - ssh_host, - ssh_port, - ssh_key_filename, init_command, unbuffered, ) @@ -275,11 +240,9 @@ def connect( FIELD_TYPE.DATE: lambda obj: convert_date(obj) or obj, }) + # todo: still needed? was conditional on SSH tunnels defer_connect = False - if ssh_host: - defer_connect = True - client_flag = pymysql.constants.CLIENT.INTERACTIVE if init_command and len(list(iocommands.split_queries(init_command))) > 1: client_flag |= pymysql.constants.CLIENT.MULTI_STATEMENTS @@ -326,26 +289,6 @@ def connect( else: raise - if ssh_host and not self.sandbox_mode: - ##### paramiko.Channel is a bad socket implementation overall if you want SSL through an SSH tunnel - ##### - # instead let's open a tunnel and rewrite host:port to local bind - try: - chan = sshtunnel.SSHTunnelForwarder( - (ssh_host, ssh_port), - ssh_username=ssh_user, - ssh_pkey=ssh_key_filename, - ssh_password=ssh_password, - remote_bind_address=(host, port), - ) - chan.start() - - conn.host = chan.local_bind_host - conn.port = chan.local_bind_port - conn.connect() - except Exception as e: - raise e - if self.conn is not None: try: self.conn.close() diff --git a/pyproject.toml b/pyproject.toml index 3165a381..2bbee4e6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,10 +43,6 @@ build-backend = "setuptools.build_meta" [project.optional-dependencies] -ssh = [ - "paramiko ~= 3.5.1", - "sshtunnel ~= 0.4.0", -] llm = [ "llm ~= 0.30.0", "pydantic_core ~= 2.41.5", # Required by llm; force a newer version @@ -54,7 +50,6 @@ llm = [ "pip == 26.*", ] all = [ - "mycli[ssh]", "mycli[llm]", ] dev = [ @@ -67,8 +62,6 @@ dev = [ "pytest-random-order ~= 1.2.0", "tox ~= 4.35.0", "pdbpp ~= 0.11.7", - "paramiko ~= 3.5.1", - "sshtunnel ~= 0.4.0", "llm ~= 0.30.0", "pydantic_core ~= 2.41.5", # Required by llm; force a newer version "setuptools == 82.*", # Required by llm commands to install models @@ -139,7 +132,7 @@ passenv = ['PYTEST_HOST', 'PYTEST_PORT', 'PYTEST_CHARSET', 'GITHUB_ACTION'] -commands = [['uv', 'pip', 'install', '-e', '.[dev,ssh,llm]'], +commands = [['uv', 'pip', 'install', '-e', '.[dev,llm]'], ['coverage', 'run', '-m', 'pytest', '-v', 'test'], ['coverage', 'report', '-m', '--sort=Miss'], ['behave', 'test/features']] @@ -153,11 +146,7 @@ commands = [['ruff', 'check'], ['ruff', 'format', '--diff']] [tool.pytest] -addopts = ['--ignore=mycli/packages/paramiko_stub/__init__.py', '--random-order'] +addopts = ['--random-order'] [tool.coverage.run] source = ['mycli'] -omit = [ - # deprecated - 'mycli/packages/paramiko_stub/__init__.py', -] diff --git a/test/pytests/conftest.py b/test/pytests/conftest.py index 7cecff4d..a62ec86a 100644 --- a/test/pytests/conftest.py +++ b/test/pytests/conftest.py @@ -3,7 +3,7 @@ import pytest import mycli.sqlexecute -from test.utils import CHARACTER_SET, DATABASE, HOST, PASSWORD, PORT, SSH_HOST, SSH_PORT, SSH_USER, USER, create_db, db_connection +from test.utils import CHARACTER_SET, DATABASE, HOST, PASSWORD, PORT, USER, create_db, db_connection @pytest.fixture(scope="function") @@ -33,9 +33,4 @@ def executor(connection): character_set=CHARACTER_SET, local_infile=False, ssl=None, - ssh_user=SSH_USER, - ssh_host=SSH_HOST, - ssh_port=SSH_PORT, - ssh_password=None, - ssh_key_filename=None, ) diff --git a/test/pytests/test_cli_runner.py b/test/pytests/test_cli_runner.py index ed9a44ca..e55600cd 100644 --- a/test/pytests/test_cli_runner.py +++ b/test/pytests/test_cli_runner.py @@ -58,7 +58,6 @@ def default_config() -> dict[str, Any]: def make_cli_args() -> main.CliArgs: cli_args = main.CliArgs() cli_args.format = None - cli_args.ssh_config_path = '/dev/null' return cli_args @@ -165,6 +164,48 @@ def test_run_from_cli_args_reports_missing_dsn(monkeypatch: pytest.MonkeyPatch) ] +def test_run_from_cli_args_rejects_unknown_positional_dsn_scheme(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args = make_cli_args() + cli_args.database = 'ssh://user@example.com/db' + secho_calls: list[tuple[str, dict[str, Any]]] = [] + monkeypatch.setattr(cli_runner.click, 'secho', lambda text, **kwargs: secho_calls.append((text, kwargs))) + + with pytest.raises(SystemExit) as excinfo: + run_with_client(monkeypatch, cli_args, DummyMyCli()) + + assert excinfo.value.code == 1 + assert secho_calls == [ + ( + 'Error: Unknown connection scheme provided for DSN URI (ssh://)', + {'err': True, 'fg': 'red'}, + ) + ] + + +def test_run_from_cli_args_rejects_unknown_alias_dsn_scheme(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args = make_cli_args() + cli_args.dsn = 'legacy_ssh' + client = DummyMyCli( + config={ + **default_config(), + 'alias_dsn': {'legacy_ssh': 'ssh://user@example.com/db'}, + } + ) + secho_calls: list[tuple[str, dict[str, Any]]] = [] + monkeypatch.setattr(cli_runner.click, 'secho', lambda text, **kwargs: secho_calls.append((text, kwargs))) + + with pytest.raises(SystemExit) as excinfo: + run_with_client(monkeypatch, cli_args, client) + + assert excinfo.value.code == 1 + assert secho_calls == [ + ( + 'Error: Unknown connection scheme provided for DSN URI (ssh://)', + {'err': True, 'fg': 'red'}, + ) + ] + + def test_run_from_cli_args_maps_dsn_ssl_parameters(monkeypatch: pytest.MonkeyPatch) -> None: cli_args = make_cli_args() cli_args.dsn = ( diff --git a/test/pytests/test_cli_utils.py b/test/pytests/test_cli_utils.py index 1d01d3e6..f0709489 100644 --- a/test/pytests/test_cli_utils.py +++ b/test/pytests/test_cli_utils.py @@ -30,7 +30,7 @@ def test_filtered_sys_argv(monkeypatch, argv, expected): ('mysqlx://user@localhost/db', True, None), ('tcp://localhost:3306', True, None), ('socket:///tmp/mysql.sock', True, None), - ('ssh://user@example.com', True, None), + ('ssh://user@example.com', False, 'ssh'), ('postgres://user@localhost/db', False, 'postgres'), ('http://example.com', False, 'http'), ], diff --git a/test/pytests/test_completion_refresher.py b/test/pytests/test_completion_refresher.py index 62cfe835..53859325 100644 --- a/test/pytests/test_completion_refresher.py +++ b/test/pytests/test_completion_refresher.py @@ -48,11 +48,6 @@ def make_sqlexecute() -> SimpleNamespace: character_set='utf8mb4', local_infile=False, ssl={'ca': 'ca.pem'}, - ssh_user='ssh-user', - ssh_host='ssh-host', - ssh_port=22, - ssh_password='ssh-pw', - ssh_key_filename='id_rsa', ) @@ -277,11 +272,6 @@ def second_callback(completer) -> None: 'utf8mb4', False, {'ca': 'ca.pem'}, - 'ssh-user', - 'ssh-host', - 22, - 'ssh-pw', - 'id_rsa', ) ] assert len(executors) == 1 diff --git a/test/pytests/test_main.py b/test/pytests/test_main.py index b16d3495..d2aab398 100644 --- a/test/pytests/test_main.py +++ b/test/pytests/test_main.py @@ -114,11 +114,6 @@ def test_binary_display_hex(executor): None, None, None, - None, - None, - None, - None, - None, ) m.explicit_pager = False sqlresult = next(m.sqlexecute.run("select b'01101010' AS binary_test")) @@ -154,11 +149,6 @@ def test_binary_display_utf8(executor): None, None, None, - None, - None, - None, - None, - None, ) m.explicit_pager = False sqlresult = next(m.sqlexecute.run("select b'01101010' AS binary_test")) @@ -386,11 +376,6 @@ def test_reconnect_database_is_selected(executor, capsys): None, None, None, - None, - None, - None, - None, - None, ) try: next(m.sqlexecute.run(f"use {DATABASE}")) @@ -421,11 +406,6 @@ def test_reconnect_no_database(executor, capsys): None, None, None, - None, - None, - None, - None, - None, ) sql = "\\r" result = next(mycli.packages.special.execute(executor, sql)) @@ -449,11 +429,6 @@ def test_reconnect_with_different_database(executor): None, None, None, - None, - None, - None, - None, - None, ) database_1 = TEST_DATABASE database_2 = DEFAULT_DATABASE @@ -480,11 +455,6 @@ def test_reconnect_with_same_database(executor): None, None, None, - None, - None, - None, - None, - None, ) database = DEFAULT_DATABASE sql = f"\\u {database}" @@ -942,35 +912,6 @@ def test_list_dsn(monkeypatch): print(f"An error occurred while attempting to delete the file: {e}") -@pytest.mark.skipif(os.name == 'nt', reason='todo: unknown') -def test_list_ssh_config(): - runner = CliRunner() - # keep Windows from locking the file with delete=False - with NamedTemporaryFile(prefix=TEMPFILE_PREFIX, mode="w", delete=False) as ssh_config: - ssh_config.write( - dedent("""\ - Host test - Hostname test.example.com - User joe - Port 22222 - IdentityFile ~/.ssh/gateway - """) - ) - ssh_config.flush() - args = ["--list-ssh-config", "--ssh-config-path", ssh_config.name] - result = runner.invoke(click_entrypoint, args=args) - assert "test\n" in result.output - result = runner.invoke(click_entrypoint, args=args + ["--verbose"]) - assert "test : test.example.com\n" in result.output - - # delete=False means we should try to clean up - try: - if os.path.exists(ssh_config.name): - os.remove(ssh_config.name) - except Exception as e: - print(f"An error occurred while attempting to delete the file: {e}") - - def test_dsn(monkeypatch): # Setup classes to mock mycli.main.MyCli class Formatter: @@ -2000,108 +1941,6 @@ def run_query(self, query, new_line=True): ) -@pytest.mark.skipif(os.name == 'nt', reason='todo: unknown') -def test_ssh_config(monkeypatch): - # Setup classes to mock mycli.main.MyCli - class Formatter: - format_name = None - - class Logger: - def debug(self, *args, **args_dict): - pass - - def warning(self, *args, **args_dict): - pass - - class MockMyCli: - config = { - "main": {}, - "alias_dsn": {}, - "connection": { - "default_keepalive_ticks": 0, - }, - } - - def __init__(self, **args): - self.logger = Logger() - self.destructive_warning = False - self.main_formatter = Formatter() - self.redirect_formatter = Formatter() - self.ssl_mode = "auto" - self.my_cnf = {"client": {}, "mysqld": {}} - self.default_keepalive_ticks = 0 - - def connect(self, **args): - MockMyCli.connect_args = args - - def run_query(self, query, new_line=True): - pass - - import mycli.main - - monkeypatch.setattr(mycli.main, "MyCli", MockMyCli) - runner = CliRunner() - - # Setup temporary configuration - # keep Windows from locking the file with delete=False - with NamedTemporaryFile(prefix=TEMPFILE_PREFIX, mode="w", delete=False) as ssh_config: - ssh_config.write( - dedent("""\ - Host test - Hostname test.example.com - User joe - Port 22222 - IdentityFile ~/.ssh/gateway - """) - ) - ssh_config.flush() - - # When a user supplies a ssh config. - result = runner.invoke(mycli.main.click_entrypoint, args=["--ssh-config-path", ssh_config.name, "--ssh-config-host", "test"]) - assert result.exit_code == 0, result.output + " " + str(result.exception) - assert ( - MockMyCli.connect_args["ssh_user"] == "joe" - and MockMyCli.connect_args["ssh_host"] == "test.example.com" - and MockMyCli.connect_args["ssh_port"] == 22222 - and MockMyCli.connect_args["ssh_key_filename"] == os.path.expanduser("~") + "/.ssh/gateway" - ) - - # When a user supplies a ssh config host as argument to mycli, - # and used command line arguments, use the command line - # arguments. - result = runner.invoke( - mycli.main.click_entrypoint, - args=[ - "--ssh-config-path", - ssh_config.name, - "--ssh-config-host", - "test", - "--ssh-user", - "arg_user", - "--ssh-host", - "arg_host", - "--ssh-port", - "3", - "--ssh-key-filename", - "/path/to/key", - ], - ) - assert result.exit_code == 0, result.output + " " + str(result.exception) - assert ( - MockMyCli.connect_args["ssh_user"] == "arg_user" - and MockMyCli.connect_args["ssh_host"] == "arg_host" - and MockMyCli.connect_args["ssh_port"] == 3 - and MockMyCli.connect_args["ssh_key_filename"] == "/path/to/key" - ) - - # delete=False means we should try to clean up - try: - if os.path.exists(ssh_config.name): - os.remove(ssh_config.name) - except Exception as e: - print(f"An error occurred while attempting to delete the file: {e}") - - @dbtest def test_init_command_arg(executor): init_command = "set sql_select_limit=1000" diff --git a/test/pytests/test_main_modes_list_ssh_config.py b/test/pytests/test_main_modes_list_ssh_config.py deleted file mode 100644 index 9ff104a4..00000000 --- a/test/pytests/test_main_modes_list_ssh_config.py +++ /dev/null @@ -1,95 +0,0 @@ -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: int = 0 - - -class DummyMyCli: - def __init__(self, config: Any) -> None: - self.config = config - self.verbosity = 0 - - -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: - mycli = DummyMyCli(config={}) - mycli.verbosity = cli_args.verbose - return list_ssh_config_mode.main_list_ssh_config(cast(Any, mycli), 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=0)) - - 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=1)) - - 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')] diff --git a/test/pytests/test_schema_prefetcher.py b/test/pytests/test_schema_prefetcher.py index 0eebe8b6..9ea81a8a 100644 --- a/test/pytests/test_schema_prefetcher.py +++ b/test/pytests/test_schema_prefetcher.py @@ -53,11 +53,6 @@ def make_mycli( character_set='utf8mb4', local_infile=False, ssl=None, - ssh_user=None, - ssh_host=None, - ssh_port=22, - ssh_password=None, - ssh_key_filename=None, databases=MagicMock(return_value=list(databases)), ) return SimpleNamespace( diff --git a/test/pytests/test_sqlexecute.py b/test/pytests/test_sqlexecute.py index 21e3b63e..660a8988 100644 --- a/test/pytests/test_sqlexecute.py +++ b/test/pytests/test_sqlexecute.py @@ -1,21 +1,17 @@ # type: ignore -import builtins from datetime import time -import importlib.util import os -from pathlib import Path -import sys from types import SimpleNamespace from prompt_toolkit.formatted_text import FormattedText import pymysql import pytest +from mycli import sqlexecute from mycli.constants import TEST_DATABASE from mycli.packages.special import iocommands from mycli.packages.sqlresult import SQLResult -import mycli.sqlexecute as sqlexecute from mycli.sqlexecute import ServerInfo, ServerSpecies, SQLExecute from test.utils import dbtest, is_expanded_output, run, set_expanded_output @@ -476,28 +472,6 @@ def test_calc_mysql_version_value_raises_for_non_numeric_parts(version_string: s ServerInfo.calc_mysql_version_value(version_string) -def test_sqlexecute_import_swallows_optional_dependency_import_errors(monkeypatch) -> None: - assert sqlexecute.__file__ is not None - original_import = builtins.__import__ - - def fake_import(name, globals=None, locals=None, fromlist=(), level=0): # noqa: A002 - if name == 'paramiko': - raise ImportError('missing optional dependency') - return original_import(name, globals, locals, fromlist, level) - - module_name = 'sqlexecute_importerror_test' - spec = importlib.util.spec_from_file_location(module_name, Path(sqlexecute.__file__)) - assert spec is not None - assert spec.loader is not None - module = importlib.util.module_from_spec(spec) - monkeypatch.setattr(builtins, '__import__', fake_import) - sys.modules[module_name] = module - try: - spec.loader.exec_module(module) - finally: - sys.modules.pop(module_name, None) - - @pytest.mark.parametrize( ('server_info', 'expected'), ( @@ -669,11 +643,6 @@ def make_executor_for_connect_tests() -> SQLExecute: executor.ssl = {'ca': '/stored/ca.pem'} executor.server_info = None executor.connection_id = None - executor.ssh_user = 'stored_ssh_user' - executor.ssh_host = None - executor.ssh_port = 22 - executor.ssh_password = 'stored_ssh_password' - executor.ssh_key_filename = '/stored/key.pem' executor.init_command = 'select 1' executor.unbuffered = False executor.sandbox_mode = False @@ -847,104 +816,6 @@ def fake_connect(**_kwargs): assert exc_info.value.args == (1045, 'access denied') -def test_connect_uses_ssh_tunnel_when_ssh_host_is_set(monkeypatch) -> None: - executor = make_executor_for_connect_tests() - executor.ssl = None - new_conn = DummyConnection(server_version='8.0.36-0ubuntu0.22.04.1') - connect_kwargs = {} - tunnel_args = {} - tunnel_started = [] - - class FakeTunnel: - def __init__( - self, - ssh_address_or_host, - ssh_username=None, - ssh_pkey=None, - ssh_password=None, - remote_bind_address=None, - ) -> None: - tunnel_args['ssh_address_or_host'] = ssh_address_or_host - tunnel_args['ssh_username'] = ssh_username - tunnel_args['ssh_pkey'] = ssh_pkey - tunnel_args['ssh_password'] = ssh_password - tunnel_args['remote_bind_address'] = remote_bind_address - self.local_bind_host = '127.0.0.1' - self.local_bind_port = 4406 - - def start(self) -> None: - tunnel_started.append(True) - - def fake_connect(**kwargs): - connect_kwargs.update(kwargs) - return new_conn - - def fake_reset_connection_id(self) -> None: - self.connection_id = 7 - - monkeypatch.setattr(sqlexecute.pymysql, 'connect', fake_connect) - monkeypatch.setattr(SQLExecute, 'reset_connection_id', fake_reset_connection_id) - monkeypatch.setattr( - sqlexecute, - 'sshtunnel', - SimpleNamespace(SSHTunnelForwarder=FakeTunnel), - raising=False, - ) - - executor.connect( - host='db.internal', - port=3308, - ssh_host='bastion.internal', - ssh_port=2222, - ssh_user='alice', - ssh_password='secret', - ssh_key_filename='/tmp/id_rsa', - ) - - assert connect_kwargs['host'] == 'db.internal' - assert connect_kwargs['port'] == 3308 - assert connect_kwargs['defer_connect'] is True - assert connect_kwargs['init_command'] == 'select 1' - assert tunnel_args['ssh_address_or_host'] == ('bastion.internal', 2222) - assert tunnel_args['ssh_username'] == 'alice' - assert tunnel_args['ssh_pkey'] == '/tmp/id_rsa' - assert tunnel_args['ssh_password'] == 'secret' - assert tunnel_args['remote_bind_address'] == ('db.internal', 3308) - assert tunnel_started == [True] - assert new_conn.host == '127.0.0.1' - assert new_conn.port == 4406 - assert new_conn.connect_calls == 1 - assert executor.conn is new_conn - assert executor.host == 'db.internal' - assert executor.port == 3308 - assert executor.connection_id == 7 - - -def test_connect_reraises_ssh_tunnel_errors(monkeypatch) -> None: - executor = make_executor_for_connect_tests() - executor.ssl = None - new_conn = DummyConnection(server_version='8.0.36-0ubuntu0.22.04.1') - - class FakeTunnel: - def __init__(self, *args, **kwargs) -> None: - self.local_bind_host = '127.0.0.1' - self.local_bind_port = 4406 - - def start(self) -> None: - raise RuntimeError('tunnel failed') - - monkeypatch.setattr(sqlexecute.pymysql, 'connect', lambda **_kwargs: new_conn) - monkeypatch.setattr( - sqlexecute, - 'sshtunnel', - SimpleNamespace(SSHTunnelForwarder=FakeTunnel), - raising=False, - ) - - with pytest.raises(RuntimeError, match='tunnel failed'): - executor.connect(ssh_host='bastion.internal') - - def test_connect_sandbox_temporarily_disables_set_character_set() -> None: original_calls = [] connect_observed_stub = [] diff --git a/test/pytests/test_ssh_utils.py b/test/pytests/test_ssh_utils.py deleted file mode 100644 index 1f26ce0b..00000000 --- a/test/pytests/test_ssh_utils.py +++ /dev/null @@ -1,90 +0,0 @@ -from __future__ import annotations - -import builtins -import importlib -from pathlib import Path -import sys -from typing import TextIO - -import pytest - -from mycli.packages import paramiko_stub, ssh_utils - - -class FakeSSHConfig: - def __init__(self, parse_error: Exception | None = None) -> None: - self.parse_error = parse_error - self.parsed_text: str | None = None - - def parse(self, handle: TextIO) -> None: - if self.parse_error is not None: - raise self.parse_error - self.parsed_text = handle.read() - - -def test_read_ssh_config_parses_and_returns_config(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: - config_path = tmp_path / 'ssh_config' - config_path.write_text('Host demo\n HostName example.com\n', encoding='utf-8') - fake_ssh_config = FakeSSHConfig() - - monkeypatch.setattr(ssh_utils.paramiko.config, 'SSHConfig', lambda: fake_ssh_config) - - result = ssh_utils.read_ssh_config(str(config_path)) - - assert result is fake_ssh_config - assert fake_ssh_config.parsed_text == 'Host demo\n HostName example.com\n' - - -def test_read_ssh_config_reports_missing_file_and_exits(monkeypatch: pytest.MonkeyPatch) -> None: - secho_calls: list[tuple[str, bool, str]] = [] - - monkeypatch.setattr( - ssh_utils.click, - 'secho', - lambda message, err, fg: secho_calls.append((message, err, fg)), - ) - - with pytest.raises(SystemExit) as excinfo: - ssh_utils.read_ssh_config('/definitely/missing/ssh_config') - - assert excinfo.value.code == 1 - assert secho_calls == [("[Errno 2] No such file or directory: '/definitely/missing/ssh_config'", True, 'red')] - - -def test_read_ssh_config_reports_parse_errors_and_exits(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: - config_path = tmp_path / 'ssh_config' - config_path.write_text('Host broken\n', encoding='utf-8') - fake_ssh_config = FakeSSHConfig(parse_error=RuntimeError('bad config')) - secho_calls: list[tuple[str, bool, str]] = [] - - monkeypatch.setattr(ssh_utils.paramiko.config, 'SSHConfig', lambda: fake_ssh_config) - monkeypatch.setattr( - ssh_utils.click, - 'secho', - lambda message, err, fg: secho_calls.append((message, err, fg)), - ) - - with pytest.raises(SystemExit) as excinfo: - ssh_utils.read_ssh_config(str(config_path)) - - assert excinfo.value.code == 1 - assert secho_calls == [(f'Could not parse SSH configuration file {config_path}:\nbad config ', True, 'red')] - - -def test_ssh_utils_falls_back_to_paramiko_stub_when_paramiko_is_unavailable(monkeypatch: pytest.MonkeyPatch) -> None: - original_import = builtins.__import__ - - def fake_import(name: str, globals_=None, locals_=None, fromlist=(), level: int = 0): - if name == 'paramiko': - raise ImportError('paramiko not installed') - return original_import(name, globals_, locals_, fromlist, level) - - monkeypatch.delitem(sys.modules, 'paramiko', raising=False) - monkeypatch.setattr(builtins, '__import__', fake_import) - - reloaded = importlib.reload(ssh_utils) - - assert reloaded.paramiko is paramiko_stub.paramiko - - monkeypatch.undo() - importlib.reload(ssh_utils) diff --git a/test/utils.py b/test/utils.py index ec78f874..7d2ffa24 100644 --- a/test/utils.py +++ b/test/utils.py @@ -35,9 +35,6 @@ HOST = os.getenv("PYTEST_HOST", DEFAULT_HOST) PORT = int(os.getenv("PYTEST_PORT", DEFAULT_PORT)) CHARACTER_SET = os.getenv("PYTEST_CHARSET", DEFAULT_CHARSET) -SSH_USER = os.getenv("PYTEST_SSH_USER", None) -SSH_HOST = os.getenv("PYTEST_SSH_HOST", None) -SSH_PORT = int(os.getenv("PYTEST_SSH_PORT", "22")) TEMPFILE_PREFIX = 'mycli_test_suite_' PYGMENTS_VERSION = Version(pygments.__version__)