fix(database): preserve ACL proof credentials (OMN-15355)

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(

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),

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",
+    }