From aaa9d9676fbaacb1c9d11b8159230de01212d197 Mon Sep 17 00:00:00 2001 From: Roland Walker Date: Mon, 9 Mar 2026 06:57:19 -0400 Subject: [PATCH] move more repeated values to constants.py * DEFAULT_CHARSET * DEFAULT_DATABASE * DEFAULT_HOST * DEFAULT_PORT * DEFAULT_USER * TEST_DATABASE All of the uses of "mysql" for the DEFAULT_DATABASE may not have been captured. The uses of the string "mysql" for the DSN scheme were not touched. We could also migrate TEST_DATABASE to the test directory and split a test/constants.py out from test/utils.py. --- changelog.md | 7 ++- mycli/constants.py | 8 ++++ mycli/main.py | 25 +++++++---- test/features/db_utils.py | 12 ++--- test/features/environment.py | 7 +-- test/features/steps/crud_database.py | 6 ++- test/test_main.py | 65 +++++++++++++++------------- test/test_sqlexecute.py | 3 +- test/utils.py | 23 ++++++---- 9 files changed, 98 insertions(+), 58 deletions(-) diff --git a/changelog.md b/changelog.md index f5cb43d0..4b02e9e9 100644 --- a/changelog.md +++ b/changelog.md @@ -3,7 +3,12 @@ Upcoming (TBD) Features --------- -* Makes short toolbar message show after initial prompt +* Make short toolbar message show after initial prompt. + + +Internal +--------- +* Migrate more repeated values to `constants.py`. 1.62.0 (2026/03/07) diff --git a/mycli/constants.py b/mycli/constants.py index eec4d037..88edaa76 100644 --- a/mycli/constants.py +++ b/mycli/constants.py @@ -2,3 +2,11 @@ REPO_URL = 'https://github.com/dbcli/mycli' DOCS_URL = f'{HOME_URL}/docs' ISSUES_URL = f'{REPO_URL}/issues' + +DEFAULT_CHARSET = 'utf8mb4' +DEFAULT_DATABASE = 'mysql' +DEFAULT_HOST = 'localhost' +DEFAULT_PORT = 3306 +DEFAULT_USER = 'root' + +TEST_DATABASE = 'mycli_test_db' diff --git a/mycli/main.py b/mycli/main.py index e58cbf16..a3ee11f3 100755 --- a/mycli/main.py +++ b/mycli/main.py @@ -68,7 +68,14 @@ from mycli.compat import WIN from mycli.completion_refresher import CompletionRefresher from mycli.config import get_mylogin_cnf_path, open_mylogin_cnf, read_config_files, str_to_bool, strip_matching_quotes, write_default_config -from mycli.constants import HOME_URL, ISSUES_URL, REPO_URL +from mycli.constants import ( + DEFAULT_CHARSET, + DEFAULT_HOST, + DEFAULT_PORT, + HOME_URL, + ISSUES_URL, + REPO_URL, +) from mycli.key_bindings import mycli_bindings from mycli.lexer import MyCliLexer from mycli.packages import special @@ -630,8 +637,8 @@ def connect( int_port = port and int(port) if not int_port: - int_port = 3306 - if not host or host == "localhost": + int_port = DEFAULT_PORT + if not host or host == DEFAULT_HOST: socket = ( socket or user_connection_config.get("default_socket") @@ -655,7 +662,7 @@ def connect( elif 'default-character-set' in cnf: character_set = cnf['default-character-set'] if not character_set: - character_set = 'utf8mb4' + character_set = DEFAULT_CHARSET # Favor whichever local_infile option is set. use_local_infile = False @@ -824,15 +831,15 @@ def _connect( # Else fall back to TCP/IP localhost socket = "" - host = "localhost" - port = 3306 + host = DEFAULT_HOST + port = DEFAULT_PORT # todo should reload the keyring identifier here instead of invalidating _connect(keyring_save_eligible=False) else: raise e else: - host = host or "localhost" - port = port or 3306 + host = host or DEFAULT_HOST + port = port or DEFAULT_PORT # could try loading the keyring again here instead of assuming nothing important changed # Bad ports give particularly daft error messages @@ -1689,7 +1696,7 @@ def get_prompt(self, string: str, _render_counter: int) -> str: elif sqlexecute.host is not None: prompt_host = sqlexecute.host else: - prompt_host = "localhost" + prompt_host = DEFAULT_HOST short_prompt_host, _, _ = prompt_host.partition('.') if re.match(r'^[\d\.]+$', short_prompt_host): short_prompt_host = prompt_host diff --git a/test/features/db_utils.py b/test/features/db_utils.py index 0d50ab63..ff649dd1 100644 --- a/test/features/db_utils.py +++ b/test/features/db_utils.py @@ -2,8 +2,10 @@ import pymysql +from mycli.constants import DEFAULT_CHARSET, DEFAULT_HOST, DEFAULT_PORT -def create_db(hostname="localhost", port=3306, username=None, password=None, dbname=None): + +def create_db(hostname=DEFAULT_HOST, port=DEFAULT_PORT, username=None, password=None, dbname=None): """Create test database. :param hostname: string @@ -15,7 +17,7 @@ def create_db(hostname="localhost", port=3306, username=None, password=None, dbn """ cn = pymysql.connect( - host=hostname, port=port, user=username, password=password, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor + host=hostname, port=port, user=username, password=password, charset=DEFAULT_CHARSET, cursorclass=pymysql.cursors.DictCursor ) with cn.cursor() as cr: @@ -45,14 +47,14 @@ def create_cn(hostname, port, password, username, dbname): user=username, password=password, db=dbname, - charset="utf8mb4", + charset=DEFAULT_CHARSET, cursorclass=pymysql.cursors.DictCursor, ) return cn -def drop_db(hostname="localhost", port=3306, username=None, password=None, dbname=None): +def drop_db(hostname=DEFAULT_HOST, port=DEFAULT_PORT, username=None, password=None, dbname=None): """Drop database. :param hostname: string @@ -68,7 +70,7 @@ def drop_db(hostname="localhost", port=3306, username=None, password=None, dbnam user=username, password=password, db=dbname, - charset="utf8mb4", + charset=DEFAULT_CHARSET, cursorclass=pymysql.cursors.DictCursor, ) diff --git a/test/features/environment.py b/test/features/environment.py index c8189631..efc78f86 100644 --- a/test/features/environment.py +++ b/test/features/environment.py @@ -9,6 +9,7 @@ import fixture_utils as fixutils import pexpect +from mycli.constants import DEFAULT_HOST, DEFAULT_PORT, DEFAULT_USER from steps.wrappers import run_cli, wait_prompt from test.utils import TEMPFILE_PREFIX @@ -54,9 +55,9 @@ def before_all(context): # Store get params from config/environment variables context.conf = { - "host": context.config.userdata.get("my_test_host", os.getenv("PYTEST_HOST", "localhost")), - "port": context.config.userdata.get("my_test_port", int(os.getenv("PYTEST_PORT", "3306"))), - "user": context.config.userdata.get("my_test_user", os.getenv("PYTEST_USER", "root")), + "host": context.config.userdata.get("my_test_host", os.getenv("PYTEST_HOST", DEFAULT_HOST)), + "port": context.config.userdata.get("my_test_port", int(os.getenv("PYTEST_PORT", DEFAULT_PORT))), + "user": context.config.userdata.get("my_test_user", os.getenv("PYTEST_USER", DEFAULT_USER)), "pass": context.config.userdata.get("my_test_pass", os.getenv("PYTEST_PASSWORD", None)), "cli_command": context.config.userdata.get("my_cli_command", None) or sys.executable + ' -c "import coverage ; coverage.process_startup(); import mycli.main; mycli.main.cli()"', diff --git a/test/features/steps/crud_database.py b/test/features/steps/crud_database.py index 01f36db1..3356a112 100644 --- a/test/features/steps/crud_database.py +++ b/test/features/steps/crud_database.py @@ -11,6 +11,8 @@ import pexpect import wrappers +from mycli.constants import DEFAULT_DATABASE + @when("we create database") def step_db_create(context): @@ -53,8 +55,8 @@ def step_db_connect_tmp(context): @when("we connect to dbserver") def step_db_connect_dbserver(context): """Send connect to database.""" - context.currentdb = "mysql" - context.cli.sendline("use mysql") + context.currentdb = DEFAULT_DATABASE + context.cli.sendline(f"use {DEFAULT_DATABASE}") @then("dbcli exits") diff --git a/test/test_main.py b/test/test_main.py index f135fe92..59762348 100644 --- a/test/test_main.py +++ b/test/test_main.py @@ -13,6 +13,13 @@ from click.testing import CliRunner from pymysql.err import OperationalError +from mycli.constants import ( + DEFAULT_DATABASE, + DEFAULT_HOST, + DEFAULT_PORT, + DEFAULT_USER, + TEST_DATABASE, +) from mycli.main import EMPTY_PASSWORD_FLAG_SENTINEL, MyCli, cli, thanks_picker from mycli.packages.parseutils import is_valid_connection_scheme import mycli.packages.special @@ -40,7 +47,7 @@ default_config_file, "--defaults-file", default_config_file, - "mycli_test_db", + TEST_DATABASE, ] @@ -137,12 +144,12 @@ def test_select_from_empty_table(executor): def test_is_valid_connection_scheme_valid(executor, capsys): - is_valid, scheme = is_valid_connection_scheme("mysql://test@localhost:3306/dev") + is_valid, scheme = is_valid_connection_scheme(f"mysql://test@{DEFAULT_HOST}:{DEFAULT_PORT}/dev") assert is_valid def test_is_valid_connection_scheme_invalid(executor, capsys): - is_valid, scheme = is_valid_connection_scheme("nope://test@localhost:3306/dev") + is_valid, scheme = is_valid_connection_scheme(f"nope://test@{DEFAULT_HOST}:{DEFAULT_PORT}/dev") assert not is_valid @@ -285,8 +292,8 @@ def test_reconnect_with_different_database(executor): None, None, ) - database_1 = "mycli_test_db" - database_2 = "mysql" + database_1 = TEST_DATABASE + database_2 = DEFAULT_DATABASE sql_1 = f"use {database_1}" sql_2 = f"\\r {database_2}" _result_1 = next(mycli.packages.special.execute(executor, sql_1)) @@ -316,7 +323,7 @@ def test_reconnect_with_same_database(executor): None, None, ) - database = "mysql" + database = DEFAULT_DATABASE sql = f"\\u {database}" result = next(mycli.packages.special.execute(executor, sql)) sql = f"\\r {database}" @@ -333,11 +340,11 @@ def test_prompt_no_host_only_socket(executor): mycli.sqlexecute.server_info = ServerInfo.from_version_string("8.0.44-0ubuntu0.24.04.1") mycli.sqlexecute.host = None mycli.sqlexecute.socket = "/var/run/mysqld/mysqld.sock" - mycli.sqlexecute.user = "root" - mycli.sqlexecute.dbname = "mysql" - mycli.sqlexecute.port = "3306" + mycli.sqlexecute.user = DEFAULT_USER + mycli.sqlexecute.dbname = DEFAULT_DATABASE + mycli.sqlexecute.port = DEFAULT_PORT prompt = mycli.get_prompt(mycli.prompt_format, 0) - assert prompt == "MySQL root@localhost:mysql> " + assert prompt == f"MySQL {DEFAULT_USER}@{DEFAULT_HOST}:{DEFAULT_DATABASE}> " @dbtest @@ -348,11 +355,11 @@ def test_prompt_socket_overrides_port(executor): mycli.sqlexecute.server_info = ServerInfo.from_version_string("8.0.44-0ubuntu0.24.04.1") mycli.sqlexecute.host = None mycli.sqlexecute.socket = "/var/run/mysqld/mysqld.sock" - mycli.sqlexecute.user = "root" - mycli.sqlexecute.dbname = "mysql" - mycli.sqlexecute.port = "3306" + mycli.sqlexecute.user = DEFAULT_USER + mycli.sqlexecute.dbname = DEFAULT_DATABASE + mycli.sqlexecute.port = DEFAULT_PORT prompt = mycli.get_prompt(mycli.prompt_format, 0) - assert prompt == "MySQL root@localhost:mysqld.sock mysql> " + assert prompt == f"MySQL {DEFAULT_USER}@{DEFAULT_HOST}:mysqld.sock {DEFAULT_DATABASE}> " @dbtest @@ -361,13 +368,13 @@ def test_prompt_socket_short_host(executor): mycli.prompt_format = "\\t \\u@\\H:\\k \\d> " mycli.sqlexecute = SQLExecute mycli.sqlexecute.server_info = ServerInfo.from_version_string("8.0.44-0ubuntu0.24.04.1") - mycli.sqlexecute.host = 'localhost.localdomain' + mycli.sqlexecute.host = f'{DEFAULT_HOST}.localdomain' mycli.sqlexecute.socket = None - mycli.sqlexecute.user = "root" - mycli.sqlexecute.dbname = "mysql" - mycli.sqlexecute.port = "3306" + mycli.sqlexecute.user = DEFAULT_USER + mycli.sqlexecute.dbname = DEFAULT_DATABASE + mycli.sqlexecute.port = DEFAULT_PORT prompt = mycli.get_prompt(mycli.prompt_format, 0) - assert prompt == "MySQL root@localhost:3306 mysql> " + assert prompt == f"MySQL {DEFAULT_USER}@{DEFAULT_HOST}:{DEFAULT_PORT} {DEFAULT_DATABASE}> " @dbtest @@ -391,11 +398,11 @@ def test_disable_show_warnings(executor): @dbtest def test_output_ddl_with_warning_and_show_warnings_enabled(executor): runner = CliRunner() - db = "mycli_test_db" + db = TEST_DATABASE table = "table_that_definitely_does_not_exist_1234" sql = f"DROP TABLE IF EXISTS {db}.{table}" result = runner.invoke(cli, args=CLI_ARGS + ["--show-warnings", "--no-warn"], input=sql) - expected = "Level\tCode\tMessage\nNote\t1051\tUnknown table 'mycli_test_db.table_that_definitely_does_not_exist_1234'\n" + expected = f"Level\tCode\tMessage\nNote\t1051\tUnknown table '{db}.table_that_definitely_does_not_exist_1234'\n" assert expected in result.output @@ -992,13 +999,13 @@ def run_query(self, query, new_line=True): result = runner.invoke( mycli.main.cli, args=[ - 'mysql://dsn_user:dsn_passwd@localhost/dsn_database?socket=mysql.sock', + f'mysql://dsn_user:dsn_passwd@{DEFAULT_HOST}/dsn_database?socket=mysql.sock', ], ) 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 MockMyCli.connect_args['host'] == 'localhost' + assert MockMyCli.connect_args['host'] == DEFAULT_HOST assert MockMyCli.connect_args['database'] == 'dsn_database' assert MockMyCli.connect_args['socket'] == 'mysql.sock' @@ -1006,13 +1013,13 @@ def run_query(self, query, new_line=True): result = runner.invoke( mycli.main.cli, args=[ - 'mysql://dsn_user:dsn_passwd@localhost/dsn_database?character_set=latin1', + f'mysql://dsn_user:dsn_passwd@{DEFAULT_HOST}/dsn_database?character_set=latin1', ], ) 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 MockMyCli.connect_args['host'] == 'localhost' + assert MockMyCli.connect_args['host'] == DEFAULT_HOST assert MockMyCli.connect_args['database'] == 'dsn_database' assert MockMyCli.connect_args['character_set'] == 'latin1' @@ -1020,14 +1027,14 @@ def run_query(self, query, new_line=True): result = runner.invoke( mycli.main.cli, args=[ - 'mysql://dsn_user:dsn_passwd@localhost/dsn_database?character_set=latin1', + f'mysql://dsn_user:dsn_passwd@{DEFAULT_HOST}/dsn_database?character_set=latin1', '--character-set=utf8mb3', ], ) 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 MockMyCli.connect_args['host'] == 'localhost' + assert MockMyCli.connect_args['host'] == DEFAULT_HOST assert MockMyCli.connect_args['database'] == 'dsn_database' assert MockMyCli.connect_args['character_set'] == 'utf8mb3' @@ -1078,9 +1085,9 @@ def run_query(self, query, new_line=True): '--user', 'user', '--host', - 'localhost', + DEFAULT_HOST, '--port', - '3306', + f'{DEFAULT_PORT}', '--database', 'database', '--password', diff --git a/test/test_sqlexecute.py b/test/test_sqlexecute.py index c57541f8..469ddaec 100644 --- a/test/test_sqlexecute.py +++ b/test/test_sqlexecute.py @@ -7,6 +7,7 @@ import pymysql import pytest +from mycli.constants import TEST_DATABASE from mycli.sqlexecute import ServerInfo, ServerSpecies from test.utils import dbtest, is_expanded_output, run, set_expanded_output @@ -125,7 +126,7 @@ def test_table_and_columns_query(executor): @dbtest def test_database_list(executor): databases = executor.databases() - assert "mycli_test_db" in databases + assert TEST_DATABASE in databases @dbtest diff --git a/test/utils.py b/test/utils.py index d30472e1..7d278f4c 100644 --- a/test/utils.py +++ b/test/utils.py @@ -9,14 +9,21 @@ import pymysql import pytest +from mycli.constants import ( + DEFAULT_CHARSET, + DEFAULT_HOST, + DEFAULT_PORT, + DEFAULT_USER, + TEST_DATABASE, +) from mycli.main import special -DATABASE = "mycli_test_db" +DATABASE = TEST_DATABASE PASSWORD = os.getenv("PYTEST_PASSWORD") -USER = os.getenv("PYTEST_USER", "root") -HOST = os.getenv("PYTEST_HOST", "localhost") -PORT = int(os.getenv("PYTEST_PORT", "3306")) -CHARACTER_SET = os.getenv("PYTEST_CHARSET", "utf8mb4") +USER = os.getenv("PYTEST_USER", DEFAULT_USER) +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")) @@ -35,14 +42,14 @@ def db_connection(dbname=None): except Exception: CAN_CONNECT_TO_DB = False -dbtest = pytest.mark.skipif(not CAN_CONNECT_TO_DB, reason="Need a mysql instance at localhost accessible by user 'root'") +dbtest = pytest.mark.skipif(not CAN_CONNECT_TO_DB, reason=f"Need a mysql instance at {DEFAULT_HOST} accessible by user '{DEFAULT_USER}'") def create_db(dbname): with db_connection().cursor() as cur: try: - cur.execute("""DROP DATABASE IF EXISTS mycli_test_db""") - cur.execute("""CREATE DATABASE mycli_test_db""") + cur.execute(f"DROP DATABASE IF EXISTS {TEST_DATABASE}") + cur.execute(f"CREATE DATABASE {TEST_DATABASE}") except Exception: pass