From 0ca217c37498ca3fba27f81f7b0e4417992a13b5 Mon Sep 17 00:00:00 2001 From: Roland Walker Date: Fri, 17 Jul 2026 16:31:05 -0400 Subject: [PATCH] configurable password sources and precedence Track the various password sources in PasswordCandidates, rather than overwriting values, and resolve the credential to use in a defined order, making explicit the precedence which was previously laid out only in a code comment. The sources, and the precedence, then easily become configurable in ~/.myclirc. Motivation: as we add more sources, such as Vault, some users may wish to disable those sources entirely. Similarly, the precedence of the environment variable may be something users may want to change. It comes before the DSN, but after the CLI literal, which is a bit wonky if you think of the DSN as a CLI literal. Followups here might be warnings on uses of insecure password sources, and some additional finesse such as skipping the system keyring in the case of taking the credentials from Vault. It could also be nice to settle all PasswordCandidates in one place, before calling connect(). --- AGENTS.md | 1 + changelog.md | 5 + mycli/TIPS | 2 + mycli/cli_runner.py | 56 +++++++---- mycli/client_connection.py | 31 +++--- mycli/main.py | 8 -- mycli/myclirc | 12 +++ mycli/password_sources.py | 73 ++++++++++++++ test/myclirc | 12 +++ test/pytests/test_cli_runner.py | 103 +++++++++++++++++-- test/pytests/test_client_connection.py | 134 ++++++++++++++++--------- test/pytests/test_main.py | 55 ++++++---- test/pytests/test_password_sources.py | 70 +++++++++++++ test/pytests/test_tabular_output.py | 5 +- 14 files changed, 444 insertions(+), 123 deletions(-) create mode 100644 mycli/password_sources.py create mode 100644 test/pytests/test_password_sources.py diff --git a/AGENTS.md b/AGENTS.md index b1b08e1c..362a0f47 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -48,6 +48,7 @@ A command line client for MySQL with auto-completion and syntax highlighting. ├── mycli/packages/sqlresult.py # the `SQLResult` dataclass for holding responses ├── mycli/packages/string_utils.py # generic string utilities ├── mycli/packages/tabular_output/ # extends cli_helper with additional output formats +├── mycli/password_sources.py # password sources and precedence ├── mycli/schema_prefetcher.py # background prefetcher for multi-schema auto-completion ├── mycli/sqlcompleter.py # offers SQL completions ├── mycli/sqlexecute.py # runs SQL queries diff --git a/changelog.md b/changelog.md index 61dfdbd9..12a8630e 100644 --- a/changelog.md +++ b/changelog.md @@ -1,6 +1,11 @@ Upcoming (TBD) ============== +Features +--------- +* Make password sources and precedence configurable. + + Bug Fixes --------- * Don't create `~/.myclirc` if the user adopted the XDG layout for myclirc. diff --git a/mycli/TIPS b/mycli/TIPS index 1a7f7b79..60398738 100644 --- a/mycli/TIPS +++ b/mycli/TIPS @@ -288,6 +288,8 @@ set editor_command = "code --wait" in ~/.myclirc to edit queries using VS Code! set up an interactive output explorer using the "[explorer]" section of ~/.myclirc! +customize password sources/precedence with "password_sources" in ~/.myclirc! + ### ### redirection ### diff --git a/mycli/cli_runner.py b/mycli/cli_runner.py index f9011505..916f2c39 100644 --- a/mycli/cli_runner.py +++ b/mycli/cli_runner.py @@ -15,6 +15,7 @@ from mycli.main_modes.execute import main_execute_from_cli from mycli.main_modes.list_dsn import main_list_dsn from mycli.packages.cli_utils import is_valid_connection_scheme +from mycli.password_sources import PasswordCandidates from mycli.vault import ( DEFAULT_VAULT_EXECUTABLE, DEFAULT_VAULT_PASSWORD_FIELD, @@ -165,10 +166,15 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non database = cli_args.dbname or cli_args.database dsn_uri = None + dsn_password: str | None = None # Treat the database argument as a DSN alias only if it matches a configured alias # todo why is port tested but not socket? - truthy_password = cli_args.password not in (None, EMPTY_PASSWORD_FLAG_SENTINEL) + truthy_password = ( + cli_args.password not in (None, EMPTY_PASSWORD_FLAG_SENTINEL) + or cli_args.password_file is not None + or os.environ.get('MYSQL_PWD') is not None + ) if ( database and "://" not in database @@ -235,9 +241,6 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non database = dsn_database if not cli_args.user and dsn_user is not None: cli_args.user = dsn_user - # todo: rationalize the behavior of empty-string passwords here - if not cli_args.password and dsn_password is not None: - cli_args.password = dsn_password if not cli_args.host: cli_args.host = dsn_host if not cli_args.port: @@ -364,25 +367,40 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non use_keyring = str_to_bool(cli_args.use_keyring) reset_keyring = False - if cli_args.password is None and cli_args.vault_secret: + password_candidates = PasswordCandidates() + if cli_args.password == EMPTY_PASSWORD_FLAG_SENTINEL: + password_candidates.add_value('prompt', cli_args.password) + elif cli_args.password is not None: + password_candidates.add_value('cli_literal', cli_args.password) + if cli_args.password_file: + password_candidates.add_loader('cli_file', lambda: main_module.get_password_from_file(cli_args.password_file)) + if os.environ.get('MYSQL_PWD') is not None: + password_candidates.add_value('environment', os.environ.get('MYSQL_PWD')) + if dsn_password is not None: + password_candidates.add_value('dsn', dsn_password) + + if cli_args.vault_secret: + vault_secret = cli_args.vault_secret vault_config = mycli.config.get('vault_beta', {}) vault_address = cli_args.vault_address or os.environ.get('VAULT_ADDR') or vault_config.get('address') or None vault_mount = cli_args.vault_mount or vault_config.get('default_mount') or None vault_field = cli_args.vault_password_field or vault_config.get('default_password_field') or DEFAULT_VAULT_PASSWORD_FIELD vault_executable = vault_config.get('vault_executable') or DEFAULT_VAULT_EXECUTABLE - try: - vault_password: str | None = get_field_from_vault( - vault_field, - cli_args.vault_secret, - executable=vault_executable, - mount=vault_mount, - address=vault_address, - ) - except VaultError as exc: - click.secho(f'Error reading password from Vault: {exc}', err=True, fg='red') - sys.exit(1) - else: - vault_password = None + + def load_vault_password() -> str | None: + try: + return get_field_from_vault( + vault_field, + vault_secret, + executable=vault_executable, + mount=vault_mount, + address=vault_address, + ) + except VaultError as exc: + click.secho(f'Error reading password from Vault: {exc}', err=True, fg='red') + sys.exit(1) + + password_candidates.add_loader('vault', load_vault_password) if cli_args.user is None and cli_args.vault_secret: vault_config = mycli.config.get('vault_beta', {}) @@ -409,7 +427,7 @@ def run_from_cli_args(cli_args: 'CliArgs', client_factory: ClientFactory) -> Non mycli.connect( database=database, user=vault_username if cli_args.user is None else cli_args.user, - passwd=vault_password if cli_args.password is None else cli_args.password, + password_candidates=password_candidates, host=cli_args.host, port=cli_args.port, socket=cli_args.socket, diff --git a/mycli/client_connection.py b/mycli/client_connection.py index 42d13049..747af556 100644 --- a/mycli/client_connection.py +++ b/mycli/client_connection.py @@ -23,6 +23,7 @@ ) from mycli.packages.filepaths import guess_socket_location from mycli.packages.special.utils import format_connection_dsn +from mycli.password_sources import PasswordCandidates from mycli.sqlexecute import SQLExecute from mycli.ssh_tunnel import SshTunnel, SshTunnelError @@ -49,7 +50,7 @@ def connect( self, database: str | None = "", user: str | None = "", - passwd: str | int | None = None, + password_candidates: PasswordCandidates | None = None, host: str | None = "", port: str | int | None = "", socket: str | None = "", @@ -86,6 +87,11 @@ def connect( if not host or host == DEFAULT_HOST: socket = socket or user_connection_config.get("default_socket") or mylogin_cnf["socket"] or guess_socket_location() + if self.config.get('main', {}).get('password_sources'): + password_source_precedence = self.config.get('main').as_list('password_sources') + else: + password_source_precedence = ['prompt'] + if ssh_jump: remote_host = host or DEFAULT_HOST remote_port = int_port or DEFAULT_PORT @@ -114,7 +120,8 @@ def connect( pass sys.exit(1) - passwd = passwd if isinstance(passwd, (str, int)) else mylogin_cnf["password"] + if password_candidates is None: + password_candidates = PasswordCandidates() if not character_set: if 'main' in self.config_without_package_defaults and 'default_character_set' in self.config_without_package_defaults['main']: @@ -167,15 +174,6 @@ def connect( if not any(v for v in ssl_config.values()): ssl_config = {} - # password hierarchy - # 1. -p / --pass/--password CLI options - # 2. --password-file CLI option - # 3. envvar (MYSQL_PWD) - # 4. DSN (mysql://user:password) - # 5. Vault - # 6. .mylogin.cnf - # 7. keyring - ssh_tunnel_field = urlquote(ssh_jump or '') if ssh_tunnel_field: ssh_tunnel_field = ':' + ssh_tunnel_field @@ -183,10 +181,13 @@ def connect( keyring_domain = 'mycli.net' keyring_retrieved_cleanly = False - if passwd is None and use_keyring and not reset_keyring: - passwd = keyring.get_password(keyring_domain, keyring_identifier) - if passwd is not None: - keyring_retrieved_cleanly = True + password_candidates.add_value('mylogin_cnf', mylogin_cnf['password']) + if use_keyring and not reset_keyring: + password_candidates.add_loader('keyring', lambda: keyring.get_password(keyring_domain, keyring_identifier)) + + selected_password = password_candidates.resolve(password_source_precedence) + passwd = selected_password.value if selected_password is not None else None + keyring_retrieved_cleanly = selected_password is not None and selected_password.source == 'keyring' # prompt for password if requested by user if passwd == EMPTY_PASSWORD_FLAG_SENTINEL: diff --git a/mycli/main.py b/mycli/main.py index f1639551..d7724a78 100755 --- a/mycli/main.py +++ b/mycli/main.py @@ -390,14 +390,6 @@ def preprocess_cli_args( cli_args.database = cli_args.password cli_args.password = EMPTY_PASSWORD_FLAG_SENTINEL - if cli_args.password is None and cli_args.password_file: - password_from_file = get_password_from_file(cli_args.password_file) - if password_from_file is not None: - cli_args.password = password_from_file - - if cli_args.password is None and os.environ.get('MYSQL_PWD') is not None: - cli_args.password = os.environ.get('MYSQL_PWD') - if cli_args.resume and not cli_args.checkpoint: click.secho('Error: --resume requires a --checkpoint file.', err=True, fg='red') sys.exit(1) diff --git a/mycli/myclirc b/mycli/myclirc index 2fa4ce23..04feeb6c 100644 --- a/mycli/myclirc +++ b/mycli/myclirc @@ -205,6 +205,18 @@ enable_pager = True # Choose a specific pager pager = 'less' +# Comma-separated sources to respect for a password, in order of precendence. +# Valid choices: +# * prompt +# * cli_literal +# * cli_file +# * environment +# * dsn +# * vault +# * mylogin_cnf +# * keyring +password_sources = prompt, cli_literal, cli_file, environment, dsn, vault, mylogin_cnf, keyring + # Whether to store and retrieve passwords from the system keyring. # * False - never use keyring # * True - always use keyring diff --git a/mycli/password_sources.py b/mycli/password_sources.py new file mode 100644 index 00000000..476a4dff --- /dev/null +++ b/mycli/password_sources.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Literal + +import click + +PasswordSource = Literal[ + 'prompt', + 'cli_literal', + 'cli_file', + 'environment', + 'dsn', + 'vault', + 'mylogin_cnf', + 'keyring', +] +PasswordValue = str | int +PasswordLoader = Callable[[], PasswordValue | None] + +KNOWN_PASSWORD_SOURCES: list[PasswordSource] = [ + 'prompt', + 'cli_literal', + 'cli_file', + 'environment', + 'dsn', + 'vault', + 'mylogin_cnf', + 'keyring', +] + + +@dataclass(frozen=True, slots=True) +class SelectedPassword: + source: PasswordSource + value: PasswordValue + + +@dataclass(frozen=True, slots=True) +class PasswordCandidate: + source: PasswordSource + value: PasswordValue | None = None + loader: PasswordLoader | None = None + + def resolve(self) -> SelectedPassword | None: + value = self.loader() if self.loader is not None else self.value + if value is None: + return None + return SelectedPassword(source=self.source, value=value) + + +class PasswordCandidates: + def __init__(self) -> None: + self._candidates: dict[PasswordSource, PasswordCandidate] = {} + + def add_value(self, source: PasswordSource, value: PasswordValue | None) -> None: + self._candidates[source] = PasswordCandidate(source=source, value=value) + + def add_loader(self, source: PasswordSource, loader: PasswordLoader) -> None: + self._candidates[source] = PasswordCandidate(source=source, loader=loader) + + def resolve(self, password_source_precedence) -> SelectedPassword | None: + for source in password_source_precedence: + if source not in KNOWN_PASSWORD_SOURCES: + click.secho(f'Skipping unknown password source: {source}.', err=True, fg='red') + continue + candidate = self._candidates.get(source) + if candidate is None: + continue + if selected := candidate.resolve(): + return selected + return None diff --git a/test/myclirc b/test/myclirc index f479a24e..360d7efa 100644 --- a/test/myclirc +++ b/test/myclirc @@ -205,6 +205,18 @@ enable_pager = True # Choose a specific pager pager = python test/features/wrappager.py ---boundary--- +# Comma-separated sources to respect for a password, in order of precendence. +# Valid choices: +# * prompt +# * cli_literal +# * cli_file +# * environment +# * dsn +# * vault +# * mylogin_cnf +# * keyring +password_sources = prompt, cli_literal, cli_file, environment, dsn, vault, mylogin_cnf, keyring + # Whether to store and retrieve passwords from the system keyring. # * False - never use keyring # * True - always use keyring diff --git a/test/pytests/test_cli_runner.py b/test/pytests/test_cli_runner.py index 119bdb28..5efba802 100644 --- a/test/pytests/test_cli_runner.py +++ b/test/pytests/test_cli_runner.py @@ -6,6 +6,9 @@ import pytest from mycli import cli_runner, main +from mycli.password_sources import ( + KNOWN_PASSWORD_SOURCES, +) class DummyLogger: @@ -77,6 +80,13 @@ def run_with_client( return client +def resolve_connect_password(connect_call: dict[str, Any]) -> tuple[str, str | int] | None: + selected = connect_call['password_candidates'].resolve(KNOWN_PASSWORD_SOURCES) + if selected is None: + return None + return selected.source, selected.value + + def test_expand_dsn_alias_env_var_returns_none() -> None: assert cli_runner.expand_dsn_alias_env_var(None, 'prod') is None @@ -161,11 +171,85 @@ def test_run_from_cli_args_treats_database_as_dsn_alias(monkeypatch: pytest.Monk assert client.dsn_alias == 'prod' connect_call = client.connect_calls[-1] assert connect_call['user'] == 'u' - assert connect_call['passwd'] == 'p' + assert resolve_connect_password(connect_call) == ('dsn', 'p') assert connect_call['host'] == 'h' assert connect_call['database'] == 'db' +def test_run_from_cli_args_password_file_prevents_positional_dsn_alias( + monkeypatch: pytest.MonkeyPatch, +) -> None: + cli_args = make_cli_args() + cli_args.database = 'prod' + cli_args.password_file = 'secret.txt' + client = DummyMyCli( + config={ + **default_config(), + 'alias_dsn': {'prod': 'mysql://u:p@h/db'}, + } + ) + password_file_calls: list[str | None] = [] + + def read_password_file(password_file: str | None) -> str: + password_file_calls.append(password_file) + return 'file-secret' + + monkeypatch.setattr(main, 'get_password_from_file', read_password_file) + + run_with_client(monkeypatch, cli_args, client) + + connect_call = client.connect_calls[-1] + assert client.dsn_alias is None + assert connect_call['database'] == 'prod' + assert resolve_connect_password(connect_call) == ('cli_file', 'file-secret') + assert password_file_calls == ['secret.txt'] + + +def test_run_from_cli_args_mysql_pwd_prevents_positional_dsn_alias( + monkeypatch: pytest.MonkeyPatch, +) -> None: + cli_args = make_cli_args() + cli_args.database = 'prod' + client = DummyMyCli( + config={ + **default_config(), + 'alias_dsn': {'prod': 'mysql://u:p@h/db'}, + } + ) + monkeypatch.setenv('MYSQL_PWD', 'environment-secret') + + run_with_client(monkeypatch, cli_args, client) + + connect_call = client.connect_calls[-1] + assert client.dsn_alias is None + assert connect_call['database'] == 'prod' + assert resolve_connect_password(connect_call) == ('environment', 'environment-secret') + + +def test_run_from_cli_args_keeps_empty_cli_password_over_dsn(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args = make_cli_args() + cli_args.dsn = 'mysql://user:dsn-secret@host/db' + cli_args.password = '' + client = DummyMyCli() + + run_with_client(monkeypatch, cli_args, client) + + assert resolve_connect_password(client.connect_calls[-1]) == ('cli_literal', '') + + +def test_run_from_cli_args_preserves_cli_password_prompt(monkeypatch: pytest.MonkeyPatch) -> None: + cli_args = make_cli_args() + cli_args.password = cli_runner.EMPTY_PASSWORD_FLAG_SENTINEL + client = DummyMyCli() + + run_with_client(monkeypatch, cli_args, client) + + assert resolve_connect_password(client.connect_calls[-1]) == ( + 'prompt', + cli_runner.EMPTY_PASSWORD_FLAG_SENTINEL, + ) + + def test_run_from_cli_args_leaves_dsn_alias_env_vars_disabled_by_default( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -210,7 +294,7 @@ def test_run_from_cli_args_expands_whole_dsn_alias_env_vars_when_enabled( run_with_client(monkeypatch, cli_args, client) assert client.connect_calls[-1]['user'] == 'env_user' - assert client.connect_calls[-1]['passwd'] == 'env_pass' + assert resolve_connect_password(client.connect_calls[-1]) == ('dsn', 'env_pass') assert client.connect_calls[-1]['host'] == 'env-host' assert client.connect_calls[-1]['port'] == 3308 assert client.connect_calls[-1]['database'] == 'env_db' @@ -429,7 +513,7 @@ def test_run_from_cli_args_accepts_mysql_plus_dsn_scheme(monkeypatch: pytest.Mon run_with_client(monkeypatch, cli_args, client) assert client.connect_calls[-1]['user'] == 'user' - assert client.connect_calls[-1]['passwd'] == 'pass' + assert resolve_connect_password(client.connect_calls[-1]) == ('dsn', 'pass') assert client.connect_calls[-1]['host'] == 'host' assert client.connect_calls[-1]['port'] == 3306 assert client.connect_calls[-1]['database'] == 'db' @@ -569,7 +653,7 @@ def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> run_with_client(monkeypatch, cli_args, client) - assert client.connect_calls[-1]['passwd'] == 'vault-secret' + assert resolve_connect_password(client.connect_calls[-1]) == ('vault', 'vault-secret') assert vault_calls == [ { 'secret': 'database/prod', @@ -681,7 +765,7 @@ def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> run_with_client(monkeypatch, cli_args, client) - assert client.connect_calls[-1]['passwd'] == 'dsn-secret' + assert resolve_connect_password(client.connect_calls[-1]) == ('dsn', 'dsn-secret') assert vault_calls == [] @@ -715,7 +799,7 @@ def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> run_with_client(monkeypatch, cli_args, client) - assert client.connect_calls[-1]['passwd'] == 'vault-secret' + assert resolve_connect_password(client.connect_calls[-1]) == ('vault', 'vault-secret') assert client.connect_calls[-1]['vault_username_from_vault'] is False assert vault_calls == [ { @@ -747,7 +831,7 @@ def fake_get_field_from_vault(field: str, secret: str, **kwargs: str | None) -> run_with_client(monkeypatch, cli_args, client) - assert client.connect_calls[-1]['passwd'] == 'vault-secret' + assert resolve_connect_password(client.connect_calls[-1]) == ('vault', 'vault-secret') assert vault_calls[-1]['address'] == 'https://vault.cli' @@ -764,11 +848,12 @@ def test_run_from_cli_args_reports_vault_error(monkeypatch: pytest.MonkeyPatch) lambda *_args, **_kwargs: (_ for _ in ()).throw(cli_runner.VaultError('permission denied')), ) + run_with_client(monkeypatch, cli_args, client) + with pytest.raises(SystemExit) as excinfo: - run_with_client(monkeypatch, cli_args, client) + resolve_connect_password(client.connect_calls[-1]) assert excinfo.value.code == 1 - assert client.connect_calls == [] assert secho_calls == [('Error reading password from Vault: permission denied', {'err': True, 'fg': 'red'})] diff --git a/test/pytests/test_client_connection.py b/test/pytests/test_client_connection.py index 8788bc74..82c3b565 100644 --- a/test/pytests/test_client_connection.py +++ b/test/pytests/test_client_connection.py @@ -6,12 +6,14 @@ from types import ModuleType, SimpleNamespace from typing import Any, Literal, cast +from configobj import ConfigObj import pymysql import pytest from mycli import client_connection from mycli.client_connection import ClientConnectionMixin from mycli.constants import DEFAULT_CHARSET, EMPTY_PASSWORD_FLAG_SENTINEL, ER_MUST_CHANGE_PASSWORD_LOGIN +from mycli.password_sources import KNOWN_PASSWORD_SOURCES, PasswordCandidates class DummyLogger: @@ -36,7 +38,7 @@ def __init__( ) -> None: self.cnf = cnf or default_cnf() self.mylogin_cnf = object() - self.config = config or {'main': {}, 'connection': {}} + self.config = ConfigObj(config) or ConfigObj({'main': {'password_sources': KNOWN_PASSWORD_SOURCES}, 'connection': {}}) self.config_without_package_defaults = config_without_package_defaults or {} self.keepalive_ticks: int | None = None self.sandbox_mode = False @@ -136,7 +138,7 @@ def test_import_swallows_missing_pwd_module(monkeypatch: pytest.MonkeyPatch) -> def test_connect_defaults_to_port_socket_and_config_character_set( monkeypatch: pytest.MonkeyPatch, ) -> None: - client = DummyClient(config={'connection': {'default_character_set': 'latin1'}, 'main': {}}) + client = DummyClient(config={'connection': {'default_character_set': 'latin1'}, 'main': {'password_sources': KNOWN_PASSWORD_SOURCES}}) monkeypatch.setenv('USER', 'env_user') monkeypatch.setattr(client_connection, 'guess_socket_location', lambda: '/tmp/mysql.sock') monkeypatch.setattr(client_connection, 'WIN', True) @@ -152,7 +154,7 @@ def test_connect_defaults_to_port_socket_and_config_character_set( def test_connect_uses_character_set_from_connection_config() -> None: - client = DummyClient(config={'main': {}, 'connection': {'default_character_set': 'utf16'}}) + client = DummyClient(config={'main': {'password_sources': KNOWN_PASSWORD_SOURCES}, 'connection': {'default_character_set': 'utf16'}}) client.connect(host='db', port=3307) @@ -162,9 +164,9 @@ def test_connect_uses_character_set_from_connection_config() -> None: def test_connect_migrates_deprecated_character_set_from_main_config( monkeypatch: pytest.MonkeyPatch, ) -> None: - config_wo = WritableConfig({'main': {'default_character_set': 'utf32'}}) + config_wo = WritableConfig({'main': {'password_sources': KNOWN_PASSWORD_SOURCES, 'default_character_set': 'utf32'}}) client = DummyClient( - config={'main': {}, 'connection': {'default_character_set': 'utf16'}}, + config={'main': {'password_sources': KNOWN_PASSWORD_SOURCES}, 'connection': {'default_character_set': 'utf16'}}, config_without_package_defaults=config_wo, ) secho_calls: list[tuple[str, dict[str, Any]]] = [] @@ -197,7 +199,7 @@ def test_connect_uses_existing_connection_character_set_when_migrating( 'connection': {'default_character_set': 'utf16'}, }) client = DummyClient( - config={'main': {}, 'connection': {'default_character_set': 'latin1'}}, + config={'main': {'password_sources': KNOWN_PASSWORD_SOURCES}, 'connection': {'default_character_set': 'latin1'}}, config_without_package_defaults=config_wo, ) secho_calls: list[tuple[str, dict[str, Any]]] = [] @@ -251,33 +253,70 @@ def fake_get_password(domain: str, identifier: str) -> str: monkeypatch.setattr(client_connection.keyring, 'get_password', fake_get_password) - client.connect(user='alice', host='db', port=3307, passwd=None, use_keyring=True) + client.connect(user='alice', host='db', port=3307, use_keyring=True) assert FakeSQLExecute.calls[-1]['password'] == 'from-keyring' assert get_password_calls == [('mycli.net', 'alice@db:3307:')] -def test_connect_prompts_for_password_sentinel(monkeypatch: pytest.MonkeyPatch) -> None: - client = DummyClient() - prompts: list[str] = [] +def test_connect_uses_mylogin_password_before_keyring(monkeypatch: pytest.MonkeyPatch) -> None: + client = DummyClient(cnf={**default_cnf(), 'password': 'from-mylogin'}) + get_password_calls: list[tuple[str, str]] = [] + monkeypatch.setattr( + client_connection.keyring, + 'get_password', + lambda domain, identifier: get_password_calls.append((domain, identifier)) or 'from-mylogin', # type: ignore[func-returns-value] + ) + monkeypatch.setattr(client_connection.keyring, 'set_password', lambda *_args: None) + + client.connect(user='alice', host='db', port=3307, use_keyring=True) + + assert FakeSQLExecute.calls[-1]['password'] == 'from-mylogin' + assert get_password_calls == [('mycli.net', 'alice@db:3307:')] + + +def test_connect_resolves_supplied_candidates_after_connection_details(monkeypatch: pytest.MonkeyPatch) -> None: + client = DummyClient(cnf={**default_cnf(), 'password': 'from-mylogin'}) + candidates = PasswordCandidates() + candidates.add_value('dsn', 'from-dsn') + keyring_calls: list[tuple[str, str]] = [] + monkeypatch.setattr( + client_connection.keyring, + 'get_password', + lambda domain, identifier: keyring_calls.append((domain, identifier)) or 'from-dsn', # type: ignore[func-returns-value] + ) + monkeypatch.setattr(client_connection.keyring, 'set_password', lambda *_args: None) - def fake_prompt(text: str, **_kwargs: Any) -> str: - prompts.append(text) - return 'prompted' + client.connect(user='alice', host='db', port=3307, password_candidates=candidates, use_keyring=True) - monkeypatch.setattr(client_connection.click, 'prompt', fake_prompt) + assert FakeSQLExecute.calls[-1]['password'] == 'from-dsn' + assert keyring_calls == [('mycli.net', 'alice@db:3307:')] - client.connect(user='alice', host='db', port=3307, passwd=EMPTY_PASSWORD_FLAG_SENTINEL) + +def test_connect_prompts_for_cli_password_candidate(monkeypatch: pytest.MonkeyPatch) -> None: + client = DummyClient() + candidates = PasswordCandidates() + candidates.add_value('prompt', EMPTY_PASSWORD_FLAG_SENTINEL) + prompts: list[str] = [] + monkeypatch.setattr( + client_connection.click, + 'prompt', + lambda text, **_kwargs: prompts.append(text) or 'prompted-secret', # type: ignore[func-returns-value] + ) + + client.connect(user='alice', host='db', port=3307, password_candidates=candidates) assert prompts == ['Enter password for alice'] - assert FakeSQLExecute.calls[-1]['password'] == 'prompted' + assert FakeSQLExecute.calls[-1]['password'] == 'prompted-secret' -def test_connect_saves_password_to_keyring(monkeypatch: pytest.MonkeyPatch) -> None: +def test_connect_saves_selected_password_to_keyring(monkeypatch: pytest.MonkeyPatch) -> None: client = DummyClient() + candidates = PasswordCandidates() + candidates.add_value('dsn', 'new-secret') set_password_calls: list[tuple[str, str, str]] = [] secho_calls: list[tuple[str, dict[str, Any]]] = [] - monkeypatch.setattr(client_connection.keyring, 'get_password', lambda domain, identifier: 'old') + monkeypatch.setattr(client_connection.keyring, 'get_password', lambda *_args: 'old-secret') monkeypatch.setattr( client_connection.keyring, 'set_password', @@ -285,43 +324,30 @@ def test_connect_saves_password_to_keyring(monkeypatch: pytest.MonkeyPatch) -> N ) monkeypatch.setattr(client_connection.click, 'secho', lambda message, **kwargs: secho_calls.append((message, kwargs))) - client.connect(user='alice', host='db', port=3307, passwd='new', use_keyring=True) + client.connect(user='alice', host='db', port=3307, password_candidates=candidates, use_keyring=True) - assert set_password_calls == [('mycli.net', 'alice@db:3307:', 'new')] + assert set_password_calls == [('mycli.net', 'alice@db:3307:', 'new-secret')] assert secho_calls == [('Password saved to the system keyring at mycli.net/alice@db:3307:', {'err': True})] -def test_connect_reports_keyring_save_errors(monkeypatch: pytest.MonkeyPatch) -> None: +def test_connect_reports_keyring_save_error(monkeypatch: pytest.MonkeyPatch) -> None: client = DummyClient() + candidates = PasswordCandidates() + candidates.add_value('dsn', 'new-secret') secho_calls: list[tuple[str, dict[str, Any]]] = [] - monkeypatch.setattr(client_connection.keyring, 'get_password', lambda domain, identifier: 'old') + monkeypatch.setattr(client_connection.keyring, 'get_password', lambda *_args: 'old-secret') - def fail_set_password(domain: str, identifier: str, password: str) -> None: + def fail_set_password(*_args: Any) -> None: raise RuntimeError('locked') monkeypatch.setattr(client_connection.keyring, 'set_password', fail_set_password) monkeypatch.setattr(client_connection.click, 'secho', lambda message, **kwargs: secho_calls.append((message, kwargs))) - client.connect(user='alice', host='db', port=3307, passwd='new', use_keyring=True) + client.connect(user='alice', host='db', port=3307, password_candidates=candidates, use_keyring=True) assert secho_calls == [('Password not saved to the system keyring: locked', {'err': True, 'fg': 'red'})] -def test_connect_does_not_save_empty_password_to_keyring(monkeypatch: pytest.MonkeyPatch) -> None: - client = DummyClient() - set_password_calls: list[tuple[str, str, str]] = [] - monkeypatch.setattr(client_connection.keyring, 'get_password', lambda domain, identifier: None) - monkeypatch.setattr( - client_connection.keyring, - 'set_password', - lambda domain, identifier, password: set_password_calls.append((domain, identifier, password)), - ) - - client.connect(user='alice', host='db', port=3307, passwd='', use_keyring=True) - - assert set_password_calls == [] - - def test_connect_uses_ssh_jump_with_remote_socket(monkeypatch: pytest.MonkeyPatch) -> None: tunnel_calls: list[dict[str, Any]] = [] @@ -360,7 +386,9 @@ def close(self) -> None: pass monkeypatch.setattr(client_connection, 'SshTunnel', FakeTunnel) - client = DummyClient(config={'main': {}, 'ssh': {'ssh_executable': '/opt/bin/ssh'}, 'connection': {}}) + client = DummyClient( + config={'main': {'password_sources': KNOWN_PASSWORD_SOURCES}, 'ssh': {'ssh_executable': '/opt/bin/ssh'}, 'connection': {}} + ) client.connect( user='alice', @@ -524,7 +552,9 @@ def close(self) -> None: pass monkeypatch.setattr(client_connection, 'SshTunnel', FakeTunnel) - client = DummyClient(config={'main': {}, 'ssh': {'ssh_options': '-o LogLevel=ERROR'}, 'connection': {}}) + client = DummyClient( + config={'main': {'password_sources': KNOWN_PASSWORD_SOURCES}, 'ssh': {'ssh_options': '-o LogLevel=ERROR'}, 'connection': {}} + ) client.connect(host='db.internal', ssh_jump='bastion', ssh_cli_options='-o Compression=yes') @@ -626,7 +656,7 @@ def test_connect_retries_without_ssl_for_auto_handshake_error() -> None: def test_connect_adds_default_ssl_ca_path() -> None: - client = DummyClient(config={'main': {}, 'connection': {'default_ssl_ca_path': '/ca/path'}}) + client = DummyClient(config={'main': {'password_sources': KNOWN_PASSWORD_SOURCES}, 'connection': {'default_ssl_ca_path': '/ca/path'}}) client.connect(host='db', port=3307, ssl={'mode': 'on'}) @@ -652,13 +682,17 @@ def test_connect_prompts_and_retries_after_access_denied_without_password( ) -> None: client = DummyClient() FakeSQLExecute.effects = [op_error(client_connection.ACCESS_DENIED_ERROR), None] - monkeypatch.setattr(client_connection.click, 'prompt', lambda *_args, **_kwargs: 'new-secret') + prompts: list[str] = [] + monkeypatch.setattr( + client_connection.click, + 'prompt', + lambda text, **_kwargs: prompts.append(text) or 'new-secret', # type: ignore[func-returns-value] + ) - client.connect(user='alice', host='db', port=3307, passwd=None) + client.connect(user='alice', host='db', port=3307) - assert len(FakeSQLExecute.calls) == 2 - assert FakeSQLExecute.calls[0]['password'] is None - assert FakeSQLExecute.calls[1]['password'] == 'new-secret' + assert prompts == ['Enter password for alice'] + assert [call['password'] for call in FakeSQLExecute.calls] == [None, 'new-secret'] def test_connect_exits_when_password_retry_fails(monkeypatch: pytest.MonkeyPatch) -> None: @@ -670,9 +704,10 @@ def test_connect_exits_when_password_retry_fails(monkeypatch: pytest.MonkeyPatch monkeypatch.setattr(client_connection.click, 'prompt', lambda *_args, **_kwargs: 'new-secret') with pytest.raises(SystemExit) as excinfo: - client.connect(user='alice', host='db', port=3307, passwd=None) + client.connect(user='alice', host='db', port=3307) assert excinfo.value.code == 1 + assert [call['password'] for call in FakeSQLExecute.calls] == [None, 'new-secret'] def test_connect_exits_when_password_retry_still_has_no_password(monkeypatch: pytest.MonkeyPatch) -> None: @@ -684,9 +719,10 @@ def test_connect_exits_when_password_retry_still_has_no_password(monkeypatch: py monkeypatch.setattr(client_connection.click, 'prompt', lambda *_args, **_kwargs: None) with pytest.raises(SystemExit) as excinfo: - client.connect(user='alice', host='db', port=3307, passwd=None) + client.connect(user='alice', host='db', port=3307) assert excinfo.value.code == 1 + assert [call['password'] for call in FakeSQLExecute.calls] == [None, None] def test_connect_reports_expired_password_login_error() -> None: diff --git a/test/pytests/test_main.py b/test/pytests/test_main.py index 51c3a1a9..48bd8eda 100644 --- a/test/pytests/test_main.py +++ b/test/pytests/test_main.py @@ -55,6 +55,7 @@ import mycli.packages.special from mycli.packages.special.main import COMMANDS as SPECIAL_COMMANDS from mycli.packages.sqlresult import SQLResult +from mycli.password_sources import KNOWN_PASSWORD_SOURCES from mycli.sqlexecute import ServerInfo, SQLExecute from mycli.types import Query from test.utils import ( @@ -97,6 +98,13 @@ CLI_ARGS = CLI_ARGS_WITHOUT_DB + [TEST_DATABASE] +def resolve_connect_password(connect_args: dict[str, Any]) -> tuple[str, str | int] | None: + selected = connect_args['password_candidates'].resolve(KNOWN_PASSWORD_SOURCES) + if selected is None: + return None + return selected.source, selected.value + + @dbtest @pytest.mark.skipif(os.name == 'nt', reason='todo: unknown; try running the test suite under winpty') def test_binary_display_hex(executor): @@ -936,7 +944,7 @@ def close(self): assert result.exit_code == 0, result.output + " " + str(result.exception) assert ( MockMyCli.connect_args["user"] == "dsn_user" - and MockMyCli.connect_args["passwd"] == "dsn_passwd" + and resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') and MockMyCli.connect_args["host"] == "dsn_host" and MockMyCli.connect_args["port"] == 1 and MockMyCli.connect_args["database"] == "dsn_database" @@ -966,7 +974,7 @@ def close(self): assert result.exit_code == 0, result.output + " " + str(result.exception) assert ( MockMyCli.connect_args["user"] == "arg_user" - and MockMyCli.connect_args["passwd"] == "arg_password" + and resolve_connect_password(MockMyCli.connect_args) == ('cli_literal', 'arg_password') and MockMyCli.connect_args["host"] == "arg_host" and MockMyCli.connect_args["port"] == 3 and MockMyCli.connect_args["database"] == "arg_database" @@ -987,7 +995,7 @@ def close(self): assert result.exit_code == 0, result.output + " " + str(result.exception) assert ( MockMyCli.connect_args["user"] == "alias_dsn_user" - and MockMyCli.connect_args["passwd"] == "alias_dsn_passwd" + and resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'alias_dsn_passwd') and MockMyCli.connect_args["host"] == "alias_dsn_host" and MockMyCli.connect_args["port"] == 4 and MockMyCli.connect_args["database"] == "alias_dsn_database" @@ -1025,7 +1033,7 @@ def close(self): assert result.exit_code == 0, result.output + " " + str(result.exception) assert ( MockMyCli.connect_args["user"] == "arg_user" - and MockMyCli.connect_args["passwd"] == "arg_password" + and resolve_connect_password(MockMyCli.connect_args) == ('cli_literal', 'arg_password') and MockMyCli.connect_args["host"] == "arg_host" and MockMyCli.connect_args["port"] == 5 and MockMyCli.connect_args["database"] == "arg_database" @@ -1036,7 +1044,7 @@ def close(self): assert result.exit_code == 0, result.output + " " + str(result.exception) assert ( MockMyCli.connect_args["user"] == "dsn_user" - and MockMyCli.connect_args["passwd"] is None + and resolve_connect_password(MockMyCli.connect_args) is None and MockMyCli.connect_args["host"] == "dsn_host" and MockMyCli.connect_args["port"] == 6 and MockMyCli.connect_args["database"] == "dsn_database" @@ -1047,7 +1055,7 @@ def close(self): assert result.exit_code == 0, result.output + " " + str(result.exception) assert ( MockMyCli.connect_args["user"] == "dsn_user" - and MockMyCli.connect_args["passwd"] == "dsn_passwd" + and resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') and MockMyCli.connect_args["host"] == "dsn_host" and MockMyCli.connect_args["port"] == 6 and MockMyCli.connect_args["database"] == "dsn_database" @@ -1083,7 +1091,7 @@ def close(self): ) assert result.exit_code == 0, result.output + ' ' + str(result.exception) assert MockMyCli.connect_args['user'] == 'dsn_user' - assert MockMyCli.connect_args['passwd'] == 'dsn_passwd' + assert resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') assert MockMyCli.connect_args['host'] == 'dsn_host' assert MockMyCli.connect_args['port'] == 6 assert MockMyCli.connect_args['database'] == 'dsn_database' @@ -1100,7 +1108,7 @@ def close(self): assert result.exit_code == 0, result.output + ' ' + str(result.exception) assert ( MockMyCli.connect_args['user'] == 'dsn_user' - and MockMyCli.connect_args['passwd'] == 'dsn_passwd' + and resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') and MockMyCli.connect_args['host'] == 'dsn_host' and MockMyCli.connect_args['port'] == 6 and MockMyCli.connect_args['database'] == 'dsn_database' @@ -1115,7 +1123,7 @@ def close(self): ) assert result.exit_code == 0, result.output + ' ' + str(result.exception) assert MockMyCli.connect_args['user'] == 'dsn_user' - assert MockMyCli.connect_args['passwd'] == 'dsn_passwd' + assert resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') assert MockMyCli.connect_args['host'] == DEFAULT_HOST assert MockMyCli.connect_args['database'] == 'dsn_database' assert MockMyCli.connect_args['socket'] == 'mysql.sock' @@ -1129,7 +1137,7 @@ def close(self): ) assert result.exit_code == 0, result.output + ' ' + str(result.exception) assert MockMyCli.connect_args['user'] == 'dsn_user' - assert MockMyCli.connect_args['passwd'] == 'dsn_passwd' + assert resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') assert MockMyCli.connect_args['host'] == DEFAULT_HOST assert MockMyCli.connect_args['database'] == 'dsn_database' assert MockMyCli.connect_args['character_set'] == 'latin1' @@ -1144,7 +1152,7 @@ def close(self): ) assert result.exit_code == 0, result.output + ' ' + str(result.exception) assert MockMyCli.connect_args['user'] == 'dsn_user' - assert MockMyCli.connect_args['passwd'] == 'dsn_passwd' + assert resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') assert MockMyCli.connect_args['host'] == DEFAULT_HOST assert MockMyCli.connect_args['database'] == 'dsn_database' assert MockMyCli.connect_args['character_set'] == 'utf8mb3' @@ -1198,7 +1206,7 @@ def close(self): assert 'DSN environment variable is deprecated' not in result.output assert ( MockMyCli.connect_args['user'] == 'dsn_user' - and MockMyCli.connect_args['passwd'] == 'dsn_passwd' + and resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') and MockMyCli.connect_args['host'] == 'dsn_host' and MockMyCli.connect_args['port'] == 7 and MockMyCli.connect_args['database'] == 'dsn_database' @@ -1263,7 +1271,7 @@ def close(self): ], ) assert result.exit_code == 0, result.output + ' ' + str(result.exception) - assert MockMyCli.connect_args['passwd'] == 'cleartext_password' + assert resolve_connect_password(MockMyCli.connect_args) == ('cli_literal', 'cleartext_password') def test_password_option_overrides_password_file_and_mysql_pwd(monkeypatch): @@ -1332,7 +1340,7 @@ def close(self): ], ) assert result.exit_code == 0, result.output + ' ' + str(result.exception) - assert MockMyCli.connect_args['passwd'] == 'option_password' + assert resolve_connect_password(MockMyCli.connect_args) == ('cli_literal', 'option_password') finally: os.remove(password_file.name) @@ -1400,7 +1408,7 @@ def close(self): ], ) assert result.exit_code == 0, result.output + ' ' + str(result.exception) - assert MockMyCli.connect_args['passwd'] == 'file_password' + assert resolve_connect_password(MockMyCli.connect_args) == ('cli_file', 'file_password') finally: os.remove(password_file.name) @@ -1810,7 +1818,7 @@ def close(self): result = runner.invoke(mycli.main.click_entrypoint, args=['prod']) assert result.exit_code == 0, result.output + ' ' + str(result.exception) assert MockMyCli.connect_args['user'] == 'env_user' - assert MockMyCli.connect_args['passwd'] is None + assert resolve_connect_password(MockMyCli.connect_args) is None assert MockMyCli.connect_args['host'] is None assert MockMyCli.connect_args['port'] is None assert MockMyCli.connect_args['database'] == 'prod' @@ -1820,7 +1828,7 @@ def close(self): assert result.exit_code == 0, result.output + ' ' + str(result.exception) assert ( MockMyCli.connect_args['user'] == 'env_user' - and MockMyCli.connect_args['passwd'] == 'dsn_passwd' + and resolve_connect_password(MockMyCli.connect_args) == ('dsn', 'dsn_passwd') and MockMyCli.connect_args['host'] == 'dsn_host' and MockMyCli.connect_args['port'] == 6 and MockMyCli.connect_args['database'] == 'dsn_database' @@ -2357,23 +2365,26 @@ def test_preprocess_cli_args_rejects_unknown_dsn_scheme(capsys: pytest.CaptureFi assert 'Unknown connection scheme provided for DSN URI (postgres://)' in capsys.readouterr().err -def test_preprocess_cli_args_reads_password_file_when_password_missing( +def test_preprocess_cli_args_preserves_password_file_when_password_missing( monkeypatch: pytest.MonkeyPatch, ) -> None: cli_args = CliArgs() cli_args.password_file = 'secret.txt' - monkeypatch.setattr(main, 'get_password_from_file', lambda password_file: f'from:{password_file}') + get_password_calls: list[str] = [] + monkeypatch.setattr(main, 'get_password_from_file', lambda password_file: get_password_calls.append(password_file)) assert preprocess_cli_args(cli_args, valid_connection_scheme) == 0 - assert cli_args.password == 'from:secret.txt' + assert cli_args.password is None + assert cli_args.password_file == 'secret.txt' + assert get_password_calls == [] -def test_preprocess_cli_args_uses_mysql_pwd_when_password_and_file_missing(monkeypatch: pytest.MonkeyPatch) -> None: +def test_preprocess_cli_args_preserves_mysql_pwd_when_password_missing(monkeypatch: pytest.MonkeyPatch) -> None: cli_args = CliArgs() monkeypatch.setenv('MYSQL_PWD', 'env-secret') assert preprocess_cli_args(cli_args, valid_connection_scheme) == 0 - assert cli_args.password == 'env-secret' + assert cli_args.password is None def test_preprocess_cli_args_prefers_existing_password_over_mysql_pwd(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/test/pytests/test_password_sources.py b/test/pytests/test_password_sources.py new file mode 100644 index 00000000..72ac9b72 --- /dev/null +++ b/test/pytests/test_password_sources.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import pytest + +from mycli.password_sources import ( + KNOWN_PASSWORD_SOURCES, + PasswordCandidates, +) + + +def test_resolve_uses_fixed_precedence_instead_of_registration_order() -> None: + candidates = PasswordCandidates() + candidates.add_value('keyring', 'keyring-secret') + candidates.add_value('dsn', 'dsn-secret') + candidates.add_value('cli_literal', 'cli-secret') + + selected = candidates.resolve(KNOWN_PASSWORD_SOURCES) + + assert selected is not None + assert selected.source == 'cli_literal' + assert selected.value == 'cli-secret' + + +def test_resolve_treats_empty_string_as_password() -> None: + candidates = PasswordCandidates() + candidates.add_value('cli_literal', '') + candidates.add_value('cli_file', 'file-secret') + + selected = candidates.resolve(KNOWN_PASSWORD_SOURCES) + + assert selected is not None + assert selected.source == 'cli_literal' + assert selected.value == '' + + +def test_resolve_skips_lower_priority_lazy_loader() -> None: + loader_calls: list[str] = [] + + def load_vault_password() -> str: + loader_calls.append('vault') + return 'vault-secret' + + candidates = PasswordCandidates() + candidates.add_loader('vault', load_vault_password) + candidates.add_value('environment', 'environment-secret') + + selected = candidates.resolve(KNOWN_PASSWORD_SOURCES) + + assert selected is not None + assert selected.source == 'environment' + assert loader_calls == [] + + +def test_resolve_falls_back_when_a_loader_returns_none() -> None: + candidates = PasswordCandidates() + candidates.add_loader('vault', lambda: None) + candidates.add_value('mylogin_cnf', 'mylogin-secret') + + selected = candidates.resolve(KNOWN_PASSWORD_SOURCES) + + assert selected is not None + assert selected.source == 'mylogin_cnf' + assert selected.value == 'mylogin-secret' + + +def test_resolve_skips_unknown_source_and_returns_none(capsys: pytest.CaptureFixture[str]) -> None: + candidates = PasswordCandidates() + + assert candidates.resolve(['unknown']) is None + assert capsys.readouterr().err == 'Skipping unknown password source: unknown.\n' diff --git a/test/pytests/test_tabular_output.py b/test/pytests/test_tabular_output.py index 5019b3fb..0a21941a 100644 --- a/test/pytests/test_tabular_output.py +++ b/test/pytests/test_tabular_output.py @@ -11,6 +11,7 @@ from mycli.main import MyCli from mycli.packages.sqlresult import SQLResult +from mycli.password_sources import PasswordCandidates from test.utils import HOST, PASSWORD, PORT, USER, dbtest default_config_file = os.path.join(os.path.dirname(__file__), "../myclirc") @@ -19,7 +20,9 @@ @pytest.fixture def mycli(): cli = MyCli() - cli.connect(None, USER, PASSWORD, HOST, PORT, None, init_command=None) + pc = PasswordCandidates() + pc.add_value('cli_literal', PASSWORD) + cli.connect(None, USER, pc, HOST, PORT, None, init_command=None) yield cli cli.sqlexecute.conn.close()