test(automation): require local PostgreSQL opt-in

apps/api/tests/integration/test_backup_restore_prod_db_regressions.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 from decimal import Decimal
 from types import SimpleNamespace
 from urllib.parse import parse_qsl, urlsplit
@@ -15,14 +16,29 @@ from sqlalchemy.ext.asyncio import create_async_engine
 
 from src.jobs import backup_restore_legacy_backfill_job as backfill
 from src.services import platform_operator_service
-from tests.integration.conftest import DEV_DB_URL
+
+
+def _explicit_local_test_database_url() -> str:
+    database_url = os.getenv("BACKUP_RESTORE_TEST_DATABASE_URL", "").strip()
+    if not database_url:
+        pytest.skip("BACKUP_RESTORE_TEST_DATABASE_URL is required for real PG tests")
+
+    parsed = urlsplit(
+        database_url.replace("postgresql+asyncpg://", "postgresql://", 1)
+    )
+    if parsed.hostname not in {None, "", "localhost", "127.0.0.1", "::1"}:
+        pytest.fail("backup/restore PG regression must target a local database")
+    database_name = parsed.path.rsplit("/", 1)[-1].strip().lower()
+    if not database_name or "prod" in database_name:
+        pytest.fail("backup/restore PG regression database name is unsafe")
+    return database_url
 
 
 @pytest.mark.asyncio
 async def test_backfill_run_inserts_satisfy_no_default_cost_column() -> None:
     """The deployed schema has cost_usd NOT NULL without a column default."""
 
-    engine = create_async_engine(DEV_DB_URL, echo=False)
+    engine = create_async_engine(_explicit_local_test_database_url(), echo=False)
     cursor_run_id = UUID("7458761b-65c6-59d2-92b4-822725ba2724")
     item_run_id = UUID("3f6e4e64-4510-57f8-bf2e-c59bb1ed72d1")
     try:
@@ -136,7 +152,7 @@ async def test_direct_readback_rewrites_ssl_and_keeps_rls_context_in_transaction
 ) -> None:
     """Run the fallback SQL on PostgreSQL behind an RLS-equivalent view."""
 
-    sqlalchemy_url = DEV_DB_URL
+    sqlalchemy_url = _explicit_local_test_database_url()
     if not any(
         key in {"ssl", "sslmode"}
         for key, _value in parse_qsl(urlsplit(sqlalchemy_url).query)

apps/api/tests/test_backup_restore_legacy_backfill_job.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import re
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
@@ -203,6 +204,26 @@ def test_cursor_never_reports_completed_when_any_item_failed() -> None:
     ) == ("blocked_accounting_drift_fail_closed", "failed")
 
 
+def test_real_pg_regression_requires_explicit_local_opt_in() -> None:
+    source_path = (
+        Path(__file__).parent
+        / "integration"
+        / "test_backup_restore_prod_db_regressions.py"
+    )
+    source = source_path.read_text(encoding="utf-8")
+
+    assert "BACKUP_RESTORE_TEST_DATABASE_URL" in source
+    assert "tests.integration.conftest" not in source
+    assert "DEV_DB_URL" not in source
+    assert "pytest.skip" in source
+    assert not re.search(
+        r"postgresql(?:\+asyncpg)?://[^/\s:@]+:[^@\s/]+@",
+        source,
+    )
+    ipv4_literals = set(re.findall(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", source))
+    assert ipv4_literals <= {"127.0.0.1"}
+
+
 @pytest.mark.asyncio
 async def test_116_rows_progress_through_two_bounded_durable_cohorts(
     monkeypatch: pytest.MonkeyPatch,