"""Unit tests for library/database_query.py. Exercises the building blocks (`_build_cmd`, `_parse_rows`, `_is_read_only`, `_redact_password_env`, `_escape_value`, `_substitute_named_args`) directly. The `main()` entrypoint is integration-tested via the role's playbook — mocking AnsibleModule stdin/exit_json end-to-end adds churn without catching additional logic bugs. """ from __future__ import annotations import importlib.util import sys import unittest import unittest.mock as mock from typing import TYPE_CHECKING from . import PROJECT_ROOT if TYPE_CHECKING: from types import ModuleType MODULE_PATH = PROJECT_ROOT / "library" / "database_query.py" sys.path.insert(0, str(PROJECT_ROOT)) def _load_database_query() -> ModuleType: spec = importlib.util.spec_from_file_location( "library_database_query", str(MODULE_PATH) ) if spec is None or spec.loader is None: raise ImportError(f"Could not load: {MODULE_PATH}") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) # type: ignore[attr-defined] return mod _DB_MOD = _load_database_query() _BASE_CONFIG: dict = { "type": "postgres", "container": "peertube-database", "username": "peertube", "name": "peertube", "password": "s3cret", } class ResolveExecTargetTests(unittest.TestCase): def test_compose_no_address_returns_container(self): self.assertEqual( _DB_MOD._resolve_exec_target(_BASE_CONFIG), "peertube-database" ) def test_compose_address_equals_container_returns_container(self): cfg = dict(_BASE_CONFIG, address="peertube-database") self.assertEqual(_DB_MOD._resolve_exec_target(cfg), "peertube-database") def test_swarm_subshell_address_resolves_to_concrete_id(self): cfg = dict( _BASE_CONFIG, container="postgres", address='"$(/usr/bin/resolve-container-id postgres postgres)"', ) fake = mock.Mock(stdout="deadbeefcafe\n") with mock.patch.object(_DB_MOD.subprocess, "run", return_value=fake) as run: self.assertEqual(_DB_MOD._resolve_exec_target(cfg), "deadbeefcafe") self.assertEqual( run.call_args.args[0], ["/usr/bin/resolve-container-id", "postgres", "postgres"], ) def test_swarm_resolver_empty_falls_back_to_container(self): cfg = dict( _BASE_CONFIG, container="postgres", address='"$(/usr/bin/resolve-container-id postgres postgres)"', ) fake = mock.Mock(stdout="\n") with mock.patch.object(_DB_MOD.subprocess, "run", return_value=fake): self.assertEqual(_DB_MOD._resolve_exec_target(cfg), "postgres") class BuildCmdPostgresTests(unittest.TestCase): def test_postgres_includes_user_db_and_error_stop(self): cmd, env = _DB_MOD._build_cmd(_BASE_CONFIG, expect_rows=False) self.assertEqual(env, {"PGPASSWORD": "s3cret"}) self.assertIn("psql", cmd) self.assertEqual(cmd[cmd.index("-U") + 1], "peertube") self.assertEqual(cmd[cmd.index("-d") + 1], "peertube") self.assertIn("ON_ERROR_STOP=1", cmd) self.assertNotIn("s3cret", cmd) def test_postgres_expect_rows_adds_tuples_only_flag(self): cmd, _ = _DB_MOD._build_cmd(_BASE_CONFIG, expect_rows=True) self.assertIn("-tA", cmd) def test_postgres_no_rows_omits_tuples_only_flag(self): cmd, _ = _DB_MOD._build_cmd(_BASE_CONFIG, expect_rows=False) self.assertNotIn("-tA", cmd) class BuildCmdMariadbTests(unittest.TestCase): def _mariadb_config(self): return dict(_BASE_CONFIG, type="mariadb", container="snipe-database") def test_mariadb_prefers_mariadb_binary_by_default(self): cmd, env = _DB_MOD._build_cmd(self._mariadb_config(), expect_rows=False) self.assertEqual(env, {"MYSQL_PWD": "s3cret"}) self.assertIn("mariadb", cmd) self.assertNotIn("mysql", cmd) self.assertEqual(cmd[cmd.index("-u") + 1], "peertube") self.assertEqual(cmd[cmd.index("-D") + 1], "peertube") self.assertIn("--batch", cmd) self.assertNotIn("s3cret", cmd) def test_mariadb_falls_back_to_mysql_when_explicitly_requested(self): cmd, _ = _DB_MOD._build_cmd( self._mariadb_config(), expect_rows=False, binary="mysql" ) self.assertIn("mysql", cmd) self.assertNotIn("mariadb", cmd) def test_mariadb_expect_rows_adds_skip_column_names(self): cmd, _ = _DB_MOD._build_cmd(self._mariadb_config(), expect_rows=True) self.assertIn("--skip-column-names", cmd) class ParseRowsTests(unittest.TestCase): def test_postgres_pipe_separated(self): rows = _DB_MOD._parse_rows("1|alice\n2|bob\n", db_type="postgres") self.assertEqual(rows, [["1", "alice"], ["2", "bob"]]) def test_mariadb_tab_separated(self): rows = _DB_MOD._parse_rows("1\talice\n2\tbob\n", db_type="mariadb") self.assertEqual(rows, [["1", "alice"], ["2", "bob"]]) def test_empty_stdout_returns_empty_list(self): self.assertEqual(_DB_MOD._parse_rows("", db_type="postgres"), []) def test_blank_line_is_skipped_no_spurious_row(self): self.assertEqual(_DB_MOD._parse_rows("\n", db_type="postgres"), []) class IsReadOnlyTests(unittest.TestCase): def test_select_is_read_only(self): self.assertTrue(_DB_MOD._is_read_only("SELECT 1;")) def test_select_with_leading_whitespace(self): self.assertTrue(_DB_MOD._is_read_only(" \n SELECT 1;\n")) def test_alter_is_not_read_only(self): self.assertFalse(_DB_MOD._is_read_only("ALTER TABLE u ADD COLUMN x INT;")) def test_insert_is_not_read_only(self): self.assertFalse(_DB_MOD._is_read_only("INSERT INTO u (x) VALUES (1);")) def test_update_is_not_read_only(self): self.assertFalse(_DB_MOD._is_read_only("UPDATE u SET x = 1;")) def test_with_cte_is_treated_as_read_only(self): self.assertTrue(_DB_MOD._is_read_only("WITH x AS (SELECT 1) SELECT * FROM x;")) class RedactPasswordEnvTests(unittest.TestCase): def test_bare_env_name_token_is_kept_intact(self): cmd = ["container", "exec", "-i", "-e", "PGPASSWORD", "db", "psql"] self.assertEqual(_DB_MOD._redact_password_env(cmd), cmd) def test_assignment_form_is_redacted(self): cmd = ["container", "exec", "-i", "-e", "PGPASSWORD=s3cret", "db", "psql"] out = _DB_MOD._redact_password_env(cmd) self.assertIn("PGPASSWORD=", out) self.assertNotIn("s3cret", " ".join(out)) def test_mariadb_mysql_pwd_assignment_redacted(self): cmd = ["container", "exec", "-e", "MYSQL_PWD=p4ss", "db", "mysql"] out = _DB_MOD._redact_password_env(cmd) self.assertIn("MYSQL_PWD=", out) self.assertNotIn("p4ss", " ".join(out)) def test_unrelated_e_envs_pass_through(self): cmd = ["container", "exec", "-e", "FOO=bar", "db", "psql"] self.assertEqual(_DB_MOD._redact_password_env(cmd), cmd) class EscapeValueTests(unittest.TestCase): def test_postgres_none_becomes_null_literal(self): self.assertEqual(_DB_MOD._escape_value(None, "postgres"), "NULL") def test_postgres_bool_becomes_capitalised_literal(self): self.assertEqual(_DB_MOD._escape_value(True, "postgres"), "TRUE") self.assertEqual(_DB_MOD._escape_value(False, "postgres"), "FALSE") def test_postgres_int_and_float_pass_through(self): self.assertEqual(_DB_MOD._escape_value(42, "postgres"), "42") self.assertEqual(_DB_MOD._escape_value(3.14, "postgres"), "3.14") def test_postgres_string_is_single_quoted(self): self.assertEqual(_DB_MOD._escape_value("alice", "postgres"), "'alice'") def test_postgres_string_doubles_inner_single_quotes(self): self.assertEqual(_DB_MOD._escape_value("O'Reilly", "postgres"), "'O''Reilly'") def test_mariadb_bool_becomes_one_or_zero(self): self.assertEqual(_DB_MOD._escape_value(True, "mariadb"), "1") self.assertEqual(_DB_MOD._escape_value(False, "mariadb"), "0") def test_mariadb_string_backslash_escapes_quote_and_backslash(self): self.assertEqual(_DB_MOD._escape_value("a'b", "mariadb"), "'a\\'b'") self.assertEqual(_DB_MOD._escape_value("a\\b", "mariadb"), "'a\\\\b'") def test_unsupported_type_raises(self): with self.assertRaises(TypeError): _DB_MOD._escape_value({"a": 1}, "postgres") with self.assertRaises(TypeError): _DB_MOD._escape_value([1, 2], "postgres") class SubstituteNamedArgsTests(unittest.TestCase): def test_no_placeholders_returns_sql_unchanged(self): sql = "SELECT 1;" self.assertEqual(_DB_MOD._substitute_named_args(sql, {}, "postgres"), sql) def test_placeholder_substituted_with_escaped_value(self): sql = "SELECT * FROM users WHERE name = %(name)s AND age = %(age)s;" out = _DB_MOD._substitute_named_args( sql, {"name": "alice", "age": 30}, "postgres" ) self.assertEqual(out, "SELECT * FROM users WHERE name = 'alice' AND age = 30;") def test_unknown_placeholder_raises_key_error(self): sql = "SELECT %(x)s;" with self.assertRaisesRegex(KeyError, r"%\(x\)s"): _DB_MOD._substitute_named_args(sql, {"y": 1}, "postgres") def test_placeholders_without_args_raises_key_error(self): sql = "SELECT %(x)s;" with self.assertRaisesRegex(KeyError, r"named_args is empty"): _DB_MOD._substitute_named_args(sql, {}, "postgres") def test_unused_args_are_silently_allowed(self): sql = "SELECT %(x)s;" out = _DB_MOD._substitute_named_args( sql, {"x": 1, "unused": "ignored"}, "postgres" ) self.assertEqual(out, "SELECT 1;") def test_quote_inside_value_does_not_break_out_of_string(self): sql = "SELECT %(s)s;" out = _DB_MOD._substitute_named_args(sql, {"s": "a'b"}, "postgres") self.assertEqual(out, "SELECT 'a''b';") def test_byte_sequence_not_a_placeholder_passes_through(self): sql = "SELECT 'a %( percent thing' AS s;" self.assertEqual(_DB_MOD._substitute_named_args(sql, {}, "postgres"), sql) class EnginesRegistryTests(unittest.TestCase): """Guard the dispatch-table shape so new engines stay declarative.""" REQUIRED_KEYS = ( "binaries", "user_flag", "db_flag", "shared_flags", "rows_flags", "row_separator", "password_env", "string_quote", "string_escape_seq", "bool_true", "bool_false", ) def test_postgres_and_mariadb_are_registered(self): self.assertIn("postgres", _DB_MOD._ENGINES) self.assertIn("mariadb", _DB_MOD._ENGINES) def test_each_engine_has_complete_dialect_definition(self): for name, spec in _DB_MOD._ENGINES.items(): with self.subTest(engine=name): for key in self.REQUIRED_KEYS: self.assertIn(key, spec, f"{name!r} missing dialect key {key!r}") def test_each_engine_binaries_is_non_empty_tuple(self): for name, spec in _DB_MOD._ENGINES.items(): with self.subTest(engine=name): self.assertIsInstance(spec["binaries"], tuple) self.assertGreater(len(spec["binaries"]), 0) def test_mariadb_prefers_mariadb_over_legacy_mysql(self): self.assertEqual(_DB_MOD._ENGINES["mariadb"]["binaries"][0], "mariadb") self.assertIn("mysql", _DB_MOD._ENGINES["mariadb"]["binaries"]) class BinaryMissingTests(unittest.TestCase): """Guard the narrow trigger for the mariadb→mysql fallback.""" def _proc(self, rc, stdout="", stderr=""): import subprocess as _sp return _sp.CompletedProcess( args=[], returncode=rc, stdout=stdout, stderr=stderr ) def test_oci_runtime_not_found_message_triggers_fallback(self): proc = self._proc( 127, stdout='OCI runtime exec failed: exec failed: unable to start container process: exec: "mariadb": executable file not found in $PATH', ) self.assertTrue(_DB_MOD._binary_missing(proc, "mariadb")) def test_rc_127_for_a_different_binary_is_not_matched(self): proc = self._proc( 127, stdout='OCI runtime exec failed: exec failed: unable to start container process: exec: "mysql": executable file not found in $PATH', ) self.assertFalse(_DB_MOD._binary_missing(proc, "mariadb")) def test_rc_127_without_not_found_string_is_not_matched(self): proc = self._proc(127, stderr="ERROR 1064 (42000): syntax error near ...") self.assertFalse(_DB_MOD._binary_missing(proc, "mariadb")) def test_nonzero_rc_other_than_127_is_not_matched(self): proc = self._proc(1, stderr="connection refused") self.assertFalse(_DB_MOD._binary_missing(proc, "mariadb")) if __name__ == "__main__": # pragma: no cover unittest.main()