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)