test: verify schema integrity and PostgreSQL transactions in CI

.github/workflows/quality-and-security.yml

@@ -25,6 +25,24 @@ jobs:
         # tests/test_supply_chain_policy.py keeps these three in agreement.
         python-version: ["3.12", "3.13", "3.14"]
 
+    # Each matrix worker owns an isolated server; fixtures create/drop only
+    # UUID databases, never the configured maintenance/application database.
+    services:
+      postgres:
+        image: postgres:18-bookworm
+        env:
+          POSTGRES_USER: scanner_test
+          POSTGRES_PASSWORD: scanner_ci_only
+          POSTGRES_DB: postgres
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U scanner_test -d postgres"
+          --health-interval 5s --health-timeout 5s --health-retries 10
+    env:
+      SCANNER_TEST_POSTGRES_ADMIN_URL: postgresql+psycopg://scanner_test:scanner_ci_only@127.0.0.1:5432/postgres
+      SCANNER_REQUIRE_POSTGRES_TESTS: "1"
+
     steps:
       - name: Check out repository
         uses: actions/checkout@v7
@@ -47,6 +65,11 @@ jobs:
         run: python -m pre_commit validate-config .pre-commit-config.yaml
 
       - name: Run tests with coverage gate
+        env:
+          # runner.temp is available at step scope, after runner allocation.
+          # The private file lets Alembic and application engines share schema
+          # without borrowing configured application data or an in-memory DB.
+          DATABASE_URL: sqlite:///${{ runner.temp }}/scanner-default.db
         run: |
           # The floor sits ~two points under the measured 89% (QUAL-005) so
           # routine PRs don't fight the gate, while a large untested addition
@@ -114,6 +137,16 @@ jobs:
       - name: Smoke-test Compose stack
         run: docker compose up --build --wait --wait-timeout 180
 
+      - name: Verify application storage in final image
+        env:
+          # Template-only CI credentials; the helper creates its own databases.
+          SCANNER_TEST_POSTGRES_ADMIN_URL: postgresql+psycopg://scanner:scanner_dev_password_change_me@postgres:5432/postgres
+        run: >-
+          docker compose run --rm -T --no-deps
+          -e SCANNER_TEST_POSTGRES_ADMIN_URL -e DATABASE_URL=sqlite://
+          -e ALERT_ENABLED=false
+          scanner-ui python -m backend.storage.verification_smoke
+
       - name: Tear down Compose stack
         if: always()
         run: docker compose down --volumes --remove-orphans

backend/storage/schema_verification.py

@@ -0,0 +1,163 @@
+"""Read-only, dialect-local schema fingerprints for migration verification.
+
+Beginner note:
+    Compare migrated and ORM-created databases on the SAME database dialect.
+    PostgreSQL rewrites expressions when storing them; reflecting both sides
+    lets its own parser normalize them without deleting meaningful SQL text.
+    This helper never repairs a schema or disposes the caller's engine.
+"""
+
+from __future__ import annotations
+
+import warnings
+from collections.abc import Mapping
+from typing import Any, cast
+
+from sqlalchemy import Boolean, inspect
+from sqlalchemy.engine import Engine
+from sqlalchemy.exc import SAWarning
+
+
+def _freeze(value: Any) -> Any:
+    """Make reflected options comparable without losing expressions or order.
+
+    Args:
+        value: SQLAlchemy reflection data, including dialect-specific clauses.
+
+    Returns:
+        An immutable value retaining mapping keys, sequence order and SQL text.
+
+    Beginner note:
+        SQL literals are case and whitespace sensitive. Do not lowercase text,
+        remove spaces or strip casts to make an unexplained discrepancy pass.
+    """
+    if isinstance(value, Mapping):
+        return tuple(sorted((key, _freeze(item)) for key, item in value.items()))
+    if isinstance(value, (list, tuple)):
+        return tuple(_freeze(item) for item in value)
+    if value is None or isinstance(value, (str, int, bool)):
+        return value
+    return str(value)
+
+
+def _sqlite_index_definitions(engine: Engine, table: str) -> frozenset[tuple[object, ...]]:
+    """Retain index key semantics that generic SQLite reflection leaves out.
+
+    Args:
+        engine: Caller-owned SQLite verification engine.
+        table: Reflected table name, quoted as an identifier before each PRAGMA.
+
+    Returns:
+        Index origins, uniqueness, partial-index flags, and ordered key/auxiliary
+        columns with their sort direction and collation. Explicit index names
+        remain significant; generated constraint-index names are excluded.
+
+    Raises:
+        sqlalchemy.exc.SQLAlchemyError: SQLite cannot return the index metadata.
+
+    Beginner note:
+        BINARY and NOCASE unique indexes accept different values even when their
+        names and columns match. ``index_xinfo`` exposes those comparison rules
+        and DESC order, which SQLAlchemy's generic reflection silently omits.
+        ``index_list`` also includes indexes backing UNIQUE and PRIMARY KEY
+        constraints. Their generated suffixes depend on declaration order, so
+        compare their actual definitions instead of incidental autoindex names.
+        See https://www.sqlite.org/pragma.html#pragma_index_xinfo.
+    """
+    quote = engine.dialect.identifier_preparer.quote_identifier
+    definitions: set[tuple[object, ...]] = set()
+    with engine.connect() as connection:
+        indexes = connection.exec_driver_sql(f"PRAGMA index_list({quote(table)})").all()
+        for index in indexes:
+            columns = connection.exec_driver_sql(f"PRAGMA index_xinfo({quote(index[1])})").all()
+            # Preserve index order and key/auxiliary status; table-column numeric
+            # offsets are incidental because the logical column name is retained.
+            keys = tuple((column[0], *column[2:]) for column in columns)
+            name = index[1] if index[3] == "c" else None
+            definitions.add((name, index[2], index[3], index[4], keys))
+    return frozenset(definitions)
+
+
+def reflect_schema(engine: Engine) -> dict[str, dict[str, object]]:
+    """Fingerprint database-enforced contracts, excluding Alembic bookkeeping.
+
+    Args:
+        engine: Caller-owned engine pointing at a verification database.
+
+    Returns:
+        Tables with columns, defaults, primary keys, checks, unique constraints,
+        complete index definitions and foreign-key targets/actions/options.
+
+    Raises:
+        sqlalchemy.exc.SQLAlchemyError: The caller's database cannot be queried
+            or its schema cannot be reflected.
+        sqlalchemy.exc.SAWarning: Index reflection is incomplete, for example
+            when the dialect cannot describe an expression index. Such a schema
+            must not be reported as matching from partial evidence.
+
+    Beginner note:
+        A matching index name says nothing about its columns or WHERE clause.
+        Index/check/unique names are retained. FK and PK names are excluded:
+        migrations name some SQLite batch-rebuilt FKs that the ORM leaves
+        unnamed; enforcement depends on ordered columns, targets and options.
+        Both sides use the same dialect/version, avoiding formatting guesses.
+    """
+    inspector = inspect(engine)
+    schema: dict[str, dict[str, object]] = {}
+    for table in inspector.get_table_names():
+        if table == "alembic_version":
+            continue
+        columns = {
+            column["name"]: (
+                str(column["type"]),
+                bool(column["nullable"]),
+                (
+                    cast(str, column["default"])[1:-1]
+                    if engine.dialect.name == "sqlite"
+                    and isinstance(column["type"], Boolean)
+                    and column.get("default") in {"'0'", "'1'"}
+                    else column.get("default")
+                ),
+                _freeze(column.get("computed")),
+                _freeze(column.get("identity")),
+            )
+            for column in inspector.get_columns(table)
+        }
+        # Retain every reflected index field: expressions, predicates, column
+        # sorting, include columns and unique flags all affect index behavior.
+        # A skipped expression index is incomplete evidence, not convergence.
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", SAWarning)
+            indexes = {_freeze(index) for index in inspector.get_indexes(table)}
+        foreign_keys = {
+            (
+                tuple(fk["constrained_columns"]),
+                fk.get("referred_schema"),
+                fk["referred_table"],
+                tuple(fk["referred_columns"]),
+                _freeze(fk.get("options", {})),
+            )
+            for fk in inspector.get_foreign_keys(table)
+        }
+        sqlite_foreign_key_actions: object = None
+        if engine.dialect.name == "sqlite":
+            # SQLite's SQL-text parser can omit inline REFERENCES actions.
+            # PRAGMA reports the actual enforcement contract independently of
+            # DDL formatting, including composite-key order (the seq column).
+            quoted_table = engine.dialect.identifier_preparer.quote_identifier(table)
+            with engine.connect() as connection:
+                rows = connection.exec_driver_sql(f"PRAGMA foreign_key_list({quoted_table})").all()
+            sqlite_foreign_key_actions = frozenset(tuple(row[1:]) for row in rows)
+        schema[table] = {
+            "columns": columns,
+            "primary_key": tuple(inspector.get_pk_constraint(table)["constrained_columns"]),
+            "checks": {_freeze(check) for check in inspector.get_check_constraints(table)},
+            "unique_constraints": {_freeze(unique) for unique in inspector.get_unique_constraints(table)},
+            "indexes": indexes,
+            "sqlite_index_definitions": (
+                _sqlite_index_definitions(engine, table) if engine.dialect.name == "sqlite" else None
+            ),
+            "foreign_keys": foreign_keys,
+            "sqlite_foreign_key_actions": sqlite_foreign_key_actions,
+        }
+    return schema

backend/storage/verification_database.py

@@ -0,0 +1,88 @@
+"""Provision verification databases without touching application schemas."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+from uuid import uuid4
+
+from sqlalchemy import create_engine
+from sqlalchemy.engine import make_url
+from sqlalchemy.exc import ArgumentError
+from sqlalchemy.pool import NullPool
+
+
+@contextmanager
+def disposable_postgres_database(admin_url: str) -> Iterator[str]:
+    """Create and finally drop a randomly named PostgreSQL database.
+
+    Args:
+        admin_url: Explicit test-server URL whose role can CREATE DATABASE.
+            Its path database is only a maintenance connection target. A query
+            ``dbname`` override is refused; TLS and other driver options remain.
+
+    Yields:
+        A URL for the newly created private database, using the pinned driver.
+
+    Raises:
+        ValueError: The URL cannot be parsed, does not name PostgreSQL, or has a
+            query option that would override the owned scratch database name.
+        RuntimeError: Engine setup, provisioning or cleanup fails; configuration
+            and credentials are suppressed from the public exception message.
+
+    Beginner note:
+        A shared ``public`` schema is never disposable. The only DROP target is
+        the locally generated UUID name whose CREATE succeeded in this context.
+        NullPool and AUTOCOMMIT are necessary: PostgreSQL CREATE/DROP DATABASE
+        cannot run in a transaction, and pooled scratch connections block DROP.
+        FORCE terminates only connections to our own scratch database on supported PostgreSQL.
+    """
+    # URL parsing and engine construction can fail before engine.connect. Keep
+    # those phases inside the same safe-error policy: parser/plugin errors may
+    # otherwise repeat opaque values from the supplied connection setting.
+    try:
+        url = make_url(admin_url)
+    except (ArgumentError, TypeError, ValueError):
+        raise ValueError("Verification provisioning requires a valid PostgreSQL URL") from None
+    if url.get_backend_name() != "postgresql":
+        raise ValueError("Verification provisioning requires a PostgreSQL URL")
+    # SQLAlchemy applies query parameters after translating the URL path into
+    # psycopg's dbname argument. Keeping a query override would route our yielded
+    # scratch URLs back into an existing database despite URL.set(database=...).
+    # Refuse it before constructing an engine or issuing any provisioning DDL.
+    if "dbname" in url.query:
+        raise ValueError("Verification URLs cannot contain a database-routing query option")
+    url = url.set(drivername="postgresql+psycopg")
+    # No caller-controlled identifiers enter these DDL strings. Even a hostile
+    # database component in admin_url cannot change the generated DROP target.
+    database_name = "scanner_verify_" + uuid4().hex
+    try:
+        engine = create_engine(
+            url, isolation_level="AUTOCOMMIT", poolclass=NullPool,
+            connect_args={"connect_timeout": 5},
+        )
+    except Exception:
+        raise RuntimeError("Cannot configure the disposable PostgreSQL verification connection") from None
+    created = False
+    try:
+        try:
+            with engine.connect() as connection:
+                connection.exec_driver_sql(f'CREATE DATABASE "{database_name}" TEMPLATE template0')
+                # CREATE is already committed. Track ownership before connection
+                # cleanup can raise so finally still drops our private database.
+                created = True
+        except Exception:
+            raise RuntimeError(
+                "Cannot provision disposable PostgreSQL database; check the explicit test URL and CREATEDB privilege"
+            ) from None
+        yield url.set(database=database_name).render_as_string(hide_password=False)
+    finally:
+        try:
+            if created:
+                try:
+                    with engine.connect() as connection:
+                        connection.exec_driver_sql(f'DROP DATABASE "{database_name}" WITH (FORCE)')
+                except Exception:
+                    raise RuntimeError(f"Cannot clean up disposable PostgreSQL database {database_name}") from None
+        finally:
+            engine.dispose()