test: add provider postgres conformance

src/user_engine/adapters/postgres.py

@@ -90,9 +90,9 @@ class PostgresUserEngineStore:
         return self.schema_version == LATEST_SCHEMA_VERSION
 
     def migrate(self) -> None:
-        sql = _load_bootstrap_sql()
         with self._cursor() as cursor:
-            cursor.execute(sql)
+            for statement in _bootstrap_sql_statements():
+                cursor.execute(statement)
         self.connection.commit()
 
     @contextmanager
@@ -549,16 +549,20 @@ class PostgresUserEngineStore:
         )
 
     def _has_latest_schema(self) -> bool:
-        with self._cursor() as cursor:
-            cursor.execute(
-                """
-                SELECT 1
-                FROM user_engine_schema_versions
-                WHERE version = %s
-                """,
-                (LATEST_SCHEMA_VERSION,),
-            )
-            return cursor.fetchone() is not None
+        try:
+            with self._cursor() as cursor:
+                cursor.execute(
+                    """
+                    SELECT 1
+                    FROM user_engine_schema_versions
+                    WHERE version = %s
+                    """,
+                    (LATEST_SCHEMA_VERSION,),
+                )
+                return cursor.fetchone() is not None
+        except Exception:
+            self.connection.rollback()
+            return False
 
     @contextmanager
     def _cursor(self) -> Iterator[PostgresCursor]:
@@ -624,3 +628,11 @@ def _load_bootstrap_sql() -> str:
     return (repo_root / "migrations/postgres/0001_user_engine_store.sql").read_text(
         encoding="utf-8"
     )
+
+
+def _bootstrap_sql_statements() -> tuple[str, ...]:
+    return tuple(
+        f"{statement.strip()};"
+        for statement in _load_bootstrap_sql().split(";")
+        if statement.strip()
+    )

src/user_engine/testing/postgres_provider.py

@@ -0,0 +1,82 @@
+"""Opt-in live Postgres conformance helpers for provider repositories."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+from user_engine.adapters.postgres import PostgresConnection, PostgresUserEngineStore
+
+POSTGRES_TEST_DSN_ENV = "USER_ENGINE_POSTGRES_TEST_DSN"
+POSTGRES_TEST_RESET_ENV = "USER_ENGINE_POSTGRES_TEST_RESET"
+_TRUTHY = {"1", "true", "yes", "on"}
+_TABLES = (
+    "user_engine_outbox_events",
+    "user_engine_audit_records",
+    "user_engine_records",
+    "user_engine_schema_versions",
+)
+
+
+@dataclass(frozen=True)
+class PostgresProviderTestConfig:
+    """Configuration for destructive provider-backed Postgres tests."""
+
+    dsn: str
+
+
+def postgres_provider_test_config(
+    environ: Mapping[str, str] | None = None,
+) -> tuple[PostgresProviderTestConfig | None, str | None]:
+    """Return live test config or a skip reason."""
+    env = environ or os.environ
+    dsn = env.get(POSTGRES_TEST_DSN_ENV, "").strip()
+    if not dsn:
+        return None, f"{POSTGRES_TEST_DSN_ENV} is not set"
+    reset_value = env.get(POSTGRES_TEST_RESET_ENV, "").strip().lower()
+    if reset_value not in _TRUTHY:
+        return (
+            None,
+            f"{POSTGRES_TEST_RESET_ENV}=1 is required because tests reset "
+            "user_engine_* tables",
+        )
+    return PostgresProviderTestConfig(dsn=dsn), None
+
+
+def connect_postgres_provider(dsn: str) -> PostgresConnection:
+    """Connect with psycopg3 or psycopg2 when a provider installs either one."""
+    try:
+        import psycopg  # type: ignore[import-not-found]
+
+        return psycopg.connect(dsn)  # type: ignore[no-any-return]
+    except ImportError:
+        pass
+
+    try:
+        import psycopg2  # type: ignore[import-not-found]
+
+        return psycopg2.connect(dsn)  # type: ignore[no-any-return]
+    except ImportError as exc:
+        raise RuntimeError("install psycopg or psycopg2 to run live tests") from exc
+
+
+def reset_user_engine_postgres_tables(connection: PostgresConnection) -> None:
+    """Create then empty user-engine tables in a dedicated provider test DB."""
+    PostgresUserEngineStore(connection).migrate()
+    cursor = connection.cursor()
+    try:
+        for table in _TABLES:
+            cursor.execute(f"DELETE FROM {table}")
+    finally:
+        close = getattr(cursor, "close", None)
+        if callable(close):
+            close()
+    connection.commit()
+
+
+def close_postgres_provider_connection(connection: Any) -> None:
+    """Close provider connections that expose a close method."""
+    close = getattr(connection, "close", None)
+    if callable(close):
+        close()