diff --git a/scripts/ci/prove_application_database_acl.py b/scripts/ci/prove_application_database_acl.py index a9c086ea39..7b4b0edde1 100644 --- a/scripts/ci/prove_application_database_acl.py +++ b/scripts/ci/prove_application_database_acl.py @@ -131,6 +131,13 @@ def _admin(dsn: str = ADMIN_DSN) -> psycopg2.extensions.connection: return connection +def _admin_dsn_for_database(database: str) -> str: + """Retarget the configured admin DSN without dropping authentication.""" + parameters = psycopg2.extensions.parse_dsn(ADMIN_DSN) + parameters["dbname"] = database + return cast("str", psycopg2.extensions.make_dsn(**parameters)) + + def _dict_rows( statement: str, parameters: Sequence[object] = (), @@ -2094,9 +2101,7 @@ def _scaffold_probe_matrix( def _scaffold_phase_proof(matrix: ModelApplicationDatabaseAclMatrix) -> None: """Prove P1 can precede object materialization without weakening rollback.""" probe = _scaffold_probe_matrix(matrix) - probe_dsn = ( - f"postgresql://postgres@{DATABASE_HOST}:{DATABASE_PORT}/acl_scaffold_probe" - ) + probe_dsn = _admin_dsn_for_database("acl_scaffold_probe") assert probe.status == "BLOCKED" assert probe.scaffold_status == "READY" rendered = render_application_database_acl_sql( diff --git a/tests/unit/runtime/test_health_config_prefetch_status.py b/tests/unit/runtime/test_health_config_prefetch_status.py index e7322eb801..c542a0429b 100644 --- a/tests/unit/runtime/test_health_config_prefetch_status.py +++ b/tests/unit/runtime/test_health_config_prefetch_status.py @@ -109,7 +109,7 @@ async def test_status_degraded_no_requirements( mock_requirements.errors = () # no extraction errors → degraded_no_requirements mock_extractor.return_value.extract_from_paths.return_value = mock_requirements - with patch(_EXTRACTOR_PATH, mock_extractor): + with patch(_LOAD_ENV_PATH), patch(_EXTRACTOR_PATH, mock_extractor): process = _make_process() await process._prefetch_config_from_infisical() @@ -146,6 +146,7 @@ async def test_status_ok_after_successful_prefetch( mock_handler_cls.return_value = mock_handler_instance with ( + patch(_LOAD_ENV_PATH), patch(_EXTRACTOR_PATH, mock_extractor), patch(_PREFETCHER_PATH, mock_prefetcher), patch(_HANDLER_PATH, mock_handler_cls), @@ -164,7 +165,10 @@ async def test_status_degraded_error_on_exception( monkeypatch.setenv("INFISICAL_ADDR", "http://localhost:8880") # Make the extractor raise to trigger the outer except - with patch(_EXTRACTOR_PATH, side_effect=RuntimeError("connection refused")): + with ( + patch(_LOAD_ENV_PATH), + patch(_EXTRACTOR_PATH, side_effect=RuntimeError("connection refused")), + ): process = _make_process() await process._prefetch_config_from_infisical() @@ -207,6 +211,7 @@ async def test_status_degraded_error_on_soft_failure( mock_handler_cls.return_value = mock_handler_instance with ( + patch(_LOAD_ENV_PATH), patch(_EXTRACTOR_PATH, mock_extractor), patch(_PREFETCHER_PATH, mock_prefetcher), patch(_HANDLER_PATH, mock_handler_cls), diff --git a/tests/unit/scripts/test_prove_application_database_acl.py b/tests/unit/scripts/test_prove_application_database_acl.py new file mode 100644 index 0000000000..f9a2990973 --- /dev/null +++ b/tests/unit/scripts/test_prove_application_database_acl.py @@ -0,0 +1,58 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Unit tests for the rebuilt PostgreSQL application ACL proof.""" + +from __future__ import annotations + +import importlib.util +import sys +from collections.abc import Callable +from pathlib import Path +from types import ModuleType +from typing import cast + +import psycopg2.extensions +import pytest + +_SCRIPT = ( + Path(__file__).parents[3] / "scripts" / "ci" / "prove_application_database_acl.py" +) + + +def _load_proof(monkeypatch: pytest.MonkeyPatch) -> ModuleType: + monkeypatch.setenv( + "ADMIN_DSN", + "postgresql://proof_admin:proof-secret@postgres:5432/omnidash_analytics" + "?application_name=acl-proof", # pragma: allowlist secret + ) + spec = importlib.util.spec_from_file_location( + "test_prove_application_database_acl_module", + _SCRIPT, + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, spec.name, module) + spec.loader.exec_module(module) + return module + + +def test_admin_dsn_for_database_preserves_authentication( + monkeypatch: pytest.MonkeyPatch, +) -> None: + proof = _load_proof(monkeypatch) + dsn_for_database = cast( + "Callable[[str], str]", + proof._admin_dsn_for_database, + ) + + parameters = psycopg2.extensions.parse_dsn(dsn_for_database("acl_scaffold_probe")) + + assert parameters == { + "application_name": "acl-proof", + "dbname": "acl_scaffold_probe", + "host": "postgres", + "password": "proof-secret", # pragma: allowlist secret + "port": "5432", + "user": "proof_admin", + }