|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from types import SimpleNamespace |
| 4 | +from typing import Any |
| 5 | + |
| 6 | +import pytest |
| 7 | + |
| 8 | +from mycli import cli_runner, main |
| 9 | + |
| 10 | + |
| 11 | +class DummyLogger: |
| 12 | + def __init__(self) -> None: |
| 13 | + self.debug_calls: list[tuple[tuple[Any, ...], dict[str, Any]]] = [] |
| 14 | + |
| 15 | + def debug(self, *args: Any, **kwargs: Any) -> None: |
| 16 | + self.debug_calls.append((args, kwargs)) |
| 17 | + |
| 18 | + |
| 19 | +class DummyMyCli: |
| 20 | + def __init__( |
| 21 | + self, |
| 22 | + *, |
| 23 | + config: dict[str, Any] | None = None, |
| 24 | + my_cnf: dict[str, Any] | None = None, |
| 25 | + config_without_package_defaults: dict[str, Any] | None = None, |
| 26 | + ) -> None: |
| 27 | + self.config = config or default_config() |
| 28 | + self.my_cnf = my_cnf or {'client': {}, 'mysqld': {}} |
| 29 | + self.config_without_package_defaults = config_without_package_defaults or {} |
| 30 | + self.default_keepalive_ticks = 5 |
| 31 | + self.ssl_mode: str | None = None |
| 32 | + self.logger = DummyLogger() |
| 33 | + self.dsn_alias: str | None = None |
| 34 | + self.connect_calls: list[dict[str, Any]] = [] |
| 35 | + self.run_cli_called = False |
| 36 | + self.close_called = False |
| 37 | + |
| 38 | + def connect(self, **kwargs: Any) -> None: |
| 39 | + self.connect_calls.append(dict(kwargs)) |
| 40 | + |
| 41 | + def run_cli(self) -> None: |
| 42 | + self.run_cli_called = True |
| 43 | + |
| 44 | + def close(self) -> None: |
| 45 | + self.close_called = True |
| 46 | + |
| 47 | + |
| 48 | +def default_config() -> dict[str, Any]: |
| 49 | + return { |
| 50 | + 'main': {'use_keyring': 'false', 'my_cnf_transition_done': 'true'}, |
| 51 | + 'connection': {'default_keepalive_ticks': 0}, |
| 52 | + 'alias_dsn': {}, |
| 53 | + 'init-commands': {}, |
| 54 | + 'alias_dsn.init-commands': {}, |
| 55 | + } |
| 56 | + |
| 57 | + |
| 58 | +def make_cli_args() -> main.CliArgs: |
| 59 | + cli_args = main.CliArgs() |
| 60 | + cli_args.format = None |
| 61 | + cli_args.ssh_config_path = '/dev/null' |
| 62 | + return cli_args |
| 63 | + |
| 64 | + |
| 65 | +def run_with_client( |
| 66 | + monkeypatch: pytest.MonkeyPatch, |
| 67 | + cli_args: main.CliArgs, |
| 68 | + client: DummyMyCli, |
| 69 | +) -> DummyMyCli: |
| 70 | + monkeypatch.setattr(main, 'preprocess_cli_args', lambda args, scheme_validator: 4) |
| 71 | + monkeypatch.setattr(cli_runner.sys, 'stdin', SimpleNamespace(isatty=lambda: True)) |
| 72 | + monkeypatch.setattr(cli_runner.sys.stderr, 'isatty', lambda: False) |
| 73 | + cli_runner.run_from_cli_args(cli_args, lambda **_kwargs: client) |
| 74 | + return client |
| 75 | + |
| 76 | + |
| 77 | +def test_run_from_cli_args_checkup_exits_zero(monkeypatch: pytest.MonkeyPatch) -> None: |
| 78 | + cli_args = make_cli_args() |
| 79 | + cli_args.checkup = True |
| 80 | + client = DummyMyCli() |
| 81 | + checkup_calls: list[DummyMyCli] = [] |
| 82 | + monkeypatch.setattr(cli_runner, 'main_checkup', lambda value: checkup_calls.append(value)) |
| 83 | + monkeypatch.setattr(main, 'preprocess_cli_args', lambda args, scheme_validator: 0) |
| 84 | + |
| 85 | + with pytest.raises(SystemExit) as excinfo: |
| 86 | + cli_runner.run_from_cli_args(cli_args, lambda **_kwargs: client) |
| 87 | + |
| 88 | + assert excinfo.value.code == 0 |
| 89 | + assert checkup_calls == [client] |
| 90 | + |
| 91 | + |
| 92 | +@pytest.mark.parametrize( |
| 93 | + ('csv', 'table', 'format_name', 'message'), |
| 94 | + ( |
| 95 | + (True, False, 'table', 'Conflicting --csv and --format arguments.'), |
| 96 | + (False, True, 'csv', 'Conflicting --table and --format arguments.'), |
| 97 | + ), |
| 98 | +) |
| 99 | +def test_run_from_cli_args_rejects_conflicting_format_flags( |
| 100 | + monkeypatch: pytest.MonkeyPatch, |
| 101 | + csv: bool, |
| 102 | + table: bool, |
| 103 | + format_name: str, |
| 104 | + message: str, |
| 105 | +) -> None: |
| 106 | + cli_args = make_cli_args() |
| 107 | + cli_args.csv = csv |
| 108 | + cli_args.table = table |
| 109 | + cli_args.format = format_name |
| 110 | + secho_calls: list[tuple[str, dict[str, Any]]] = [] |
| 111 | + monkeypatch.setattr(cli_runner.click, 'secho', lambda text, **kwargs: secho_calls.append((text, kwargs))) |
| 112 | + monkeypatch.setattr(main, 'preprocess_cli_args', lambda args, scheme_validator: 0) |
| 113 | + |
| 114 | + with pytest.raises(SystemExit) as excinfo: |
| 115 | + cli_runner.run_from_cli_args(cli_args, lambda **_kwargs: DummyMyCli()) |
| 116 | + |
| 117 | + assert excinfo.value.code == 1 |
| 118 | + assert secho_calls == [(message, {'err': True, 'fg': 'red'})] |
| 119 | + |
| 120 | + |
| 121 | +def test_run_from_cli_args_uses_deprecated_mysql_unix_port_and_database_alias( |
| 122 | + monkeypatch: pytest.MonkeyPatch, |
| 123 | +) -> None: |
| 124 | + cli_args = make_cli_args() |
| 125 | + cli_args.database = 'prod' |
| 126 | + client = DummyMyCli( |
| 127 | + config={ |
| 128 | + **default_config(), |
| 129 | + 'alias_dsn': {'prod': 'mysql://dsn_user:dsn_pass@dsn_host:3307/dsn_db'}, |
| 130 | + } |
| 131 | + ) |
| 132 | + secho_calls: list[str] = [] |
| 133 | + monkeypatch.setenv('MYSQL_UNIX_PORT', '/tmp/mysql.sock') |
| 134 | + monkeypatch.setattr(cli_runner.click, 'secho', lambda text, **_kwargs: secho_calls.append(text)) |
| 135 | + |
| 136 | + run_with_client(monkeypatch, cli_args, client) |
| 137 | + |
| 138 | + assert client.dsn_alias == 'prod' |
| 139 | + assert client.connect_calls[-1]['database'] == 'dsn_db' |
| 140 | + assert client.connect_calls[-1]['user'] == 'dsn_user' |
| 141 | + assert client.connect_calls[-1]['passwd'] == 'dsn_pass' |
| 142 | + assert client.connect_calls[-1]['host'] == 'dsn_host' |
| 143 | + assert client.connect_calls[-1]['port'] == 3307 |
| 144 | + assert client.connect_calls[-1]['socket'] == '/tmp/mysql.sock' |
| 145 | + assert any('MYSQL_UNIX_PORT environment variable is deprecated' in call for call in secho_calls) |
| 146 | + |
| 147 | + |
| 148 | +def test_run_from_cli_args_reports_missing_dsn(monkeypatch: pytest.MonkeyPatch) -> None: |
| 149 | + cli_args = make_cli_args() |
| 150 | + cli_args.dsn = 'missing' |
| 151 | + secho_calls: list[tuple[str, dict[str, Any]]] = [] |
| 152 | + monkeypatch.setattr(cli_runner, 'is_valid_connection_scheme', lambda value: (False, None)) |
| 153 | + monkeypatch.setattr(cli_runner.click, 'secho', lambda text, **kwargs: secho_calls.append((text, kwargs))) |
| 154 | + monkeypatch.setattr(main, 'preprocess_cli_args', lambda args, scheme_validator: 0) |
| 155 | + |
| 156 | + with pytest.raises(SystemExit) as excinfo: |
| 157 | + cli_runner.run_from_cli_args(cli_args, lambda **_kwargs: DummyMyCli()) |
| 158 | + |
| 159 | + assert excinfo.value.code == 1 |
| 160 | + assert secho_calls == [ |
| 161 | + ( |
| 162 | + 'Could not find the specified DSN in the config file. Please check the "[alias_dsn]" section in your myclirc.', |
| 163 | + {'err': True, 'fg': 'red'}, |
| 164 | + ) |
| 165 | + ] |
| 166 | + |
| 167 | + |
| 168 | +def test_run_from_cli_args_maps_dsn_ssl_parameters(monkeypatch: pytest.MonkeyPatch) -> None: |
| 169 | + cli_args = make_cli_args() |
| 170 | + cli_args.dsn = ( |
| 171 | + 'mysql://user:pass@host:3306/db?ssl=true&ssl_ca=~/ca.pem&ssl_capath=/capath' |
| 172 | + '&ssl_cert=~/cert.pem&ssl_key=~/key.pem&ssl_cipher=AES256&tls_version=TLSv1.3' |
| 173 | + '&ssl_verify_server_cert=true' |
| 174 | + ) |
| 175 | + client = DummyMyCli() |
| 176 | + secho_calls: list[str] = [] |
| 177 | + monkeypatch.setattr(cli_runner.click, 'secho', lambda text, **_kwargs: secho_calls.append(text)) |
| 178 | + |
| 179 | + run_with_client(monkeypatch, cli_args, client) |
| 180 | + |
| 181 | + ssl = client.connect_calls[-1]['ssl'] |
| 182 | + assert ssl == { |
| 183 | + 'mode': 'on', |
| 184 | + 'ca': cli_runner.os.path.expanduser('~/ca.pem'), |
| 185 | + 'capath': '/capath', |
| 186 | + 'cert': cli_runner.os.path.expanduser('~/cert.pem'), |
| 187 | + 'key': cli_runner.os.path.expanduser('~/key.pem'), |
| 188 | + 'cipher': 'AES256', |
| 189 | + 'tls_version': 'TLSv1.3', |
| 190 | + 'check_hostname': True, |
| 191 | + } |
| 192 | + assert any('"ssl" DSN URI parameter is deprecated' in call for call in secho_calls) |
| 193 | + |
| 194 | + |
| 195 | +def test_run_from_cli_args_merges_global_list_and_alias_scalar_init_commands( |
| 196 | + monkeypatch: pytest.MonkeyPatch, |
| 197 | +) -> None: |
| 198 | + cli_args = make_cli_args() |
| 199 | + cli_args.dsn = 'prod' |
| 200 | + cli_args.init_command = 'set cli=1' |
| 201 | + client = DummyMyCli( |
| 202 | + config={ |
| 203 | + **default_config(), |
| 204 | + 'alias_dsn': {'prod': 'mysql://u:p@h/db'}, |
| 205 | + 'init-commands': {'first': ['set global=1', 'set global=2']}, |
| 206 | + 'alias_dsn.init-commands': {'prod': 'set alias=1'}, |
| 207 | + } |
| 208 | + ) |
| 209 | + |
| 210 | + run_with_client(monkeypatch, cli_args, client) |
| 211 | + |
| 212 | + assert client.connect_calls[-1]['init_command'] == 'set global=1; set global=2; set alias=1; set cli=1' |
| 213 | + |
| 214 | + |
| 215 | +def test_run_from_cli_args_resets_keyring(monkeypatch: pytest.MonkeyPatch) -> None: |
| 216 | + cli_args = make_cli_args() |
| 217 | + cli_args.use_keyring = 'reset' |
| 218 | + client = DummyMyCli() |
| 219 | + |
| 220 | + run_with_client(monkeypatch, cli_args, client) |
| 221 | + |
| 222 | + assert client.connect_calls[-1]['use_keyring'] is True |
| 223 | + assert client.connect_calls[-1]['reset_keyring'] is True |
| 224 | + |
| 225 | + |
| 226 | +def test_run_from_cli_args_uses_explicit_keyring_flag(monkeypatch: pytest.MonkeyPatch) -> None: |
| 227 | + cli_args = make_cli_args() |
| 228 | + cli_args.use_keyring = 'true' |
| 229 | + client = DummyMyCli() |
| 230 | + |
| 231 | + run_with_client(monkeypatch, cli_args, client) |
| 232 | + |
| 233 | + assert client.connect_calls[-1]['use_keyring'] is True |
| 234 | + assert client.connect_calls[-1]['reset_keyring'] is False |
0 commit comments