From def72e22d6355f394aa8575073a4159828910580 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Mon, 28 Sep 2026 16:47:36 +0530 Subject: [PATCH 1/5] Strengthen schema and disposable database verification Compare database enforcement contracts and prove drift sensitivity. Exercise seeded history upgrades and repository transactions against private databases, and require PostgreSQL plus final-image storage smoke in CI. Co-authored-by: Codex --- .github/workflows/quality-and-security.yml | 29 ++ backend/storage/schema_verification.py | 115 ++++++++ backend/storage/verification_database.py | 67 +++++ backend/storage/verification_smoke.py | 274 ++++++++++++++++++ .../components/deployment-runtime.md | 35 +++ .../components/storage-persistence.md | 50 ++++ tests/conftest.py | 42 +++ tests/test_database_verification_smoke.py | 200 +++++++++++++ tests/test_scan_storage_migrations.py | 36 +-- tests/test_schema_verification.py | 139 +++++++++ tests/test_supply_chain_policy.py | 26 ++ 11 files changed, 987 insertions(+), 26 deletions(-) create mode 100644 backend/storage/schema_verification.py create mode 100644 backend/storage/verification_database.py create mode 100644 backend/storage/verification_smoke.py create mode 100644 tests/test_database_verification_smoke.py create mode 100644 tests/test_schema_verification.py diff --git a/.github/workflows/quality-and-security.yml b/.github/workflows/quality-and-security.yml index dd9a193..0af10e4 100644 --- a/.github/workflows/quality-and-security.yml +++ b/.github/workflows/quality-and-security.yml @@ -23,6 +23,25 @@ jobs: # drift before an interpreter upgrade is forced on us. python-version: ["3.11", "3.12"] + # Each matrix worker owns an isolated server; fixtures create/drop only + # UUID databases, never the configured maintenance/application database. + services: + postgres: + image: postgres:16-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: + DATABASE_URL: sqlite:// + 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@v4 @@ -112,6 +131,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 diff --git a/backend/storage/schema_verification.py b/backend/storage/schema_verification.py new file mode 100644 index 0000000..a6053d5 --- /dev/null +++ b/backend/storage/schema_verification.py @@ -0,0 +1,115 @@ +"""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 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. + + 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, + "foreign_keys": foreign_keys, + "sqlite_foreign_key_actions": sqlite_foreign_key_actions, + } + return schema diff --git a/backend/storage/verification_database.py b/backend/storage/verification_database.py new file mode 100644 index 0000000..e90e17f --- /dev/null +++ b/backend/storage/verification_database.py @@ -0,0 +1,67 @@ +"""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.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 configured database is only a maintenance connection target. + + Yields: + A URL for the newly created private database, using the pinned driver. + + Raises: + ValueError: The URL is not PostgreSQL. + RuntimeError: Provisioning or cleanup fails; credentials are suppressed. + + 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 = make_url(admin_url) + if url.get_backend_name() != "postgresql": + raise ValueError("Verification provisioning requires a PostgreSQL URL") + 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 + engine = create_engine( + url, isolation_level="AUTOCOMMIT", poolclass=NullPool, + connect_args={"connect_timeout": 5}, + ) + created = False + try: + try: + with engine.connect() as connection: + connection.exec_driver_sql(f'CREATE DATABASE "{database_name}" TEMPLATE template0') + 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() diff --git a/backend/storage/verification_smoke.py b/backend/storage/verification_smoke.py new file mode 100644 index 0000000..6aa5852 --- /dev/null +++ b/backend/storage/verification_smoke.py @@ -0,0 +1,274 @@ +"""Provider-free migration/repository verification inside the application image. + +Beginner note: + This is an explicit verification command, not an application startup hook. + It requires a separate PostgreSQL provisioning URL and owns disposable + databases and a private DATA_DIR; it never scans or sends notifications. +""" + +from __future__ import annotations + +import datetime as dt +import json +import os +import subprocess # nosec B404 +import sys +from collections.abc import Iterator +from contextlib import contextmanager +from decimal import Decimal +from pathlib import Path +from tempfile import TemporaryDirectory + +from alembic import command +from alembic.config import Config +from alembic.script import ScriptDirectory +from sqlalchemy import delete, func, inspect, select, text +from sqlalchemy.engine import Engine +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from backend.storage.schema_verification import reflect_schema +from backend.storage.verification_database import disposable_postgres_database + + +@contextmanager +def _migration_url(url: str) -> Iterator[None]: + """Temporarily direct Alembic's environment reader at a private database. + + Beginner note: + Alembic reads DATABASE_URL at invocation time. Restore the original + value even on failure so an integration test cannot redirect later tests. + This helper is synchronous; callers must not run it concurrently. + """ + original = os.environ.get("DATABASE_URL") + os.environ["DATABASE_URL"] = url + try: + yield + finally: + if original is None: + os.environ.pop("DATABASE_URL", None) + else: + os.environ["DATABASE_URL"] = original + + +def migration_config() -> Config: + """Return a config anchored to the installed application migration directory. + + Beginner note: + A programmatic Config preserves the application's logging/redaction + setup. Resolving from __file__ also works from a non-repository cwd. + """ + config = Config() + config.set_main_option("script_location", str(Path(__file__).resolve().parents[2] / "migrations")) + return config + + +def _seed_history(engine: Engine) -> tuple[int, int]: + """Insert initial-revision rows using only columns that existed then. + + Beginner note: + Current ORM INSERTs cannot seed an old schema: they mention columns + which do not exist yet. Bound SQL uses the real historical shape and + lets both dialects generate IDs so sequence state remains realistic. + """ + with engine.begin() as connection: + run_id = connection.execute(text( + "INSERT INTO scan_runs (started_at, finished_at, status, screener_key, universe_key) " + "VALUES (:started, :started, 'success', 'verification_legacy', 'nifty_100') RETURNING id" + ), {"started": dt.datetime(2026, 6, 4, tzinfo=dt.UTC).isoformat()}).scalar_one() + result_id = connection.execute(text( + "INSERT INTO scan_results (run_id, symbol, close_price, raw_result_json, created_at) " + "VALUES (:run_id, 'LEGACY', 123.4500, :payload, :created) RETURNING id" + ), {"run_id": run_id, "payload": '{"symbol":"LEGACY","history_marker":"retain"}', + "created": dt.datetime(2026, 6, 4, tzinfo=dt.UTC).isoformat()}).scalar_one() + return int(run_id), int(result_id) + + +def _verify_repository(engine: Engine, legacy_run_id: int, legacy_result_id: int) -> None: + """Prove durable repository writes, rollback, uniqueness and database cascade. + + Beginner note: + Separate transactions ensure reads observe committed storage rather than + a Session identity map. A Core parent DELETE bypasses ORM cascades, so + missing database ON DELETE CASCADE really fails this check. All queries + remain inside storage; the smoke command owns the transaction lifetime. + """ + from backend.storage import repository + from backend.storage.models import ScanResult, ScanRun, ScanStatus, UserRole + + with Session(engine) as session: + legacy = repository.get_scan_run(session, legacy_run_id) + if legacy is None or legacy.status != ScanStatus.SUCCESS: + raise AssertionError("Historical scan header was lost or changed") + rows = repository.get_scan_results(session, legacy_run_id) + if len(rows) != 1 or rows[0].id != legacy_result_id: + raise AssertionError("Historical result identity was lost") + payload = rows[0].raw_result_json + if rows[0].close_price != Decimal("123.4500") or not payload or payload.get("history_marker") != "retain": + raise AssertionError("Historical price or JSON evidence was changed") + with Session(engine) as session, session.begin(): + run = repository.create_scan_run(session, screener_key="verification_current", universe_key="nifty_100") + run_id = run.id + repository.save_scan_results(session, run, [{"symbol": "SMOKE", "close": Decimal("456.7800")}]) + repository.finish_scan_run(session, run, status=ScanStatus.SUCCESS) + repository.set_user_role(session, email="smoke@example.invalid", role="admin", assigned_by=None) + with Session(engine) as session: + stored = repository.get_scan_run(session, run_id) + if stored is None or stored.status != ScanStatus.SUCCESS: + raise AssertionError("Repository write/read failed") + if repository.get_scan_results(session, run_id)[0].close_price != Decimal("456.7800"): + raise AssertionError("Repository price precision changed") + # Direct duplicate INSERT proves actual database uniqueness instead of + # relying on the repository's friendly upsert to avoid a duplicate itself. + with Session(engine) as session: + session.add(UserRole(email="smoke@example.invalid", role="admin", assigned_by=None)) + try: + session.commit() + except IntegrityError: + session.rollback() + else: + raise AssertionError("Database accepted a duplicate user-role identity") + # A deliberate exception proves caller-owned rollback, with readback from + # a fresh transaction; failed writes must not create phantom scan history. + with Session(engine) as session: + rolled_back = repository.create_scan_run( + session, screener_key="verification_rollback", universe_key="nifty_100" + ) + rolled_back_id = rolled_back.id + session.rollback() + with Session(engine) as session, session.begin(): + if repository.get_scan_run(session, rolled_back_id) is not None: + raise AssertionError("Rolled-back scan survived") + session.execute(delete(ScanRun).where(ScanRun.id == run_id)) + if session.scalar(select(func.count()).select_from(ScanResult).where(ScanResult.run_id == run_id)) != 0: + raise AssertionError("Database did not cascade result deletion") + if repository.get_scan_run(session, legacy_run_id) is None: + raise AssertionError("Deleting the scratch run deleted unrelated history") + if repository.delete_user_role(session, "smoke@example.invalid") != "admin": + raise AssertionError("Repository delete failed") + + +def verify_database_pair(migrated_url: str, metadata_url: str) -> dict[str, bool]: + """Upgrade seeded history and compare with a separate ORM-created database. + + Args: + migrated_url: Empty private database to build through the migration chain. + metadata_url: Different empty private database to build from ORM metadata. + + Returns: + Three proof flags only after every assertion succeeds. + + Raises: + AssertionError: Database identity, history, repository, or schema differs. + + Beginner note: + Refuse nonempty databases BEFORE any DDL. Provisioning and eventual + deletion belong to the caller, and this function never drops a schema. + Comparison is within one dialect, not a lossy SQLite/PostgreSQL mapping. + """ + from backend.storage.database import _make_engine + from backend.storage.models import Base + + if migrated_url == metadata_url: + raise AssertionError("Verification requires two different private databases") + migrated, metadata = _make_engine(migrated_url), _make_engine(metadata_url) + try: + if migrated.dialect.name != metadata.dialect.name: + raise AssertionError("Verification databases must use the same dialect") + if inspect(migrated).get_table_names() or inspect(metadata).get_table_names(): + raise AssertionError("Verification databases must be empty") + config = migration_config() + if len(ScriptDirectory.from_config(config).get_heads()) != 1: + raise AssertionError("The migration chain must have exactly one head") + with _migration_url(migrated_url): + command.upgrade(config, "20260604scan002") + legacy_run_id, legacy_result_id = _seed_history(migrated) + command.upgrade(config, "head") + _verify_repository(migrated, legacy_run_id, legacy_result_id) + Base.metadata.create_all(metadata) + actual, expected = reflect_schema(migrated), reflect_schema(metadata) + differences = [ + f"{table}.{part}" + for table in actual.keys() | expected.keys() + for part in actual.get(table, {}).keys() | expected.get(table, {}).keys() + if actual.get(table, {}).get(part) != expected.get(table, {}).get(part) + ] + if differences: + raise AssertionError("Migration/ORM schema drift: " + ", ".join(sorted(differences))) + return {"history_preserved": True, "repository_round_trip": True, "schema_converged": True} + finally: + migrated.dispose() + metadata.dispose() + + +def verify_relocated_universe_sources() -> None: + """Read packaged source universes after moving mutable data to an empty dir. + + Beginner note: + PR116 separates curated source CSVs from generated DATA_DIR files. This + check must read the package's sources even when the private runtime + directory contains no universes. It performs no downloads or mappings. + """ + from backend.config import DATA_DIR + from backend.universe_builder import HEMANT_SOURCE_FILES, load_symbol_list_csv + + for source in HEMANT_SOURCE_FILES.values(): + if source.resolve().is_relative_to(DATA_DIR.resolve()): + raise AssertionError("Curated universe source incorrectly follows relocated DATA_DIR") + if not load_symbol_list_csv(source): + raise AssertionError("Packaged curated universe source is empty") + + +def run_image_verification() -> None: + """Verify image sources/storage after the launcher sets private environment. + + Beginner note: + Run in the launcher's fresh process so import-time path constants read + the relocated DATA_DIR. Suppress driver errors to avoid echoing a DSN. + Assertion messages contain only our own fixed verification diagnostics. + """ + try: + verify_relocated_universe_sources() + admin_url = os.environ["SCANNER_TEST_POSTGRES_ADMIN_URL"] + with ( + disposable_postgres_database(admin_url) as migrated, + disposable_postgres_database(admin_url) as metadata, + ): + result = verify_database_pair(migrated, metadata) + print(json.dumps({**result, "relocated_universe_sources": True}, sort_keys=True)) + except AssertionError as exc: + raise SystemExit(str(exc)) from None + except Exception as exc: + raise SystemExit(f"Application image verification failed ({type(exc).__name__})") from None + + +def main() -> int: + """Run image verification using explicitly authorized scratch resources. + + Beginner note: + The package creates its normal lazy engine at import time but does not + connect to it. The child replaces DATABASE_URL with harmless SQLite + before config import; PostgreSQL writes use only generated databases. + Failure messages omit connection URLs and raw database exceptions. + """ + admin_url = os.environ.get("SCANNER_TEST_POSTGRES_ADMIN_URL", "").strip() + if not admin_url: + raise SystemExit("SCANNER_TEST_POSTGRES_ADMIN_URL is required") + # backend.storage imports config before module execution. A fresh child + # receives DATA_DIR before that import, matching real image startup. + with TemporaryDirectory(prefix="scanner-image-verification-") as data_dir: + env = os.environ.copy() + env.update(DATABASE_URL="sqlite://", DATA_DIR=data_dir, ALERT_ENABLED="false") + # Fixed argv uses this interpreter without a shell; URLs stay in env. + completed = subprocess.run( # nosec B603 + [sys.executable, "-c", "from backend.storage.verification_smoke import run_image_verification; " + "run_image_verification()"], + env=env, check=False, timeout=180, + ) + if completed.returncode: + raise SystemExit(completed.returncode) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/architecture/components/deployment-runtime.md b/docs/architecture/components/deployment-runtime.md index 8917149..165bda9 100644 --- a/docs/architecture/components/deployment-runtime.md +++ b/docs/architecture/components/deployment-runtime.md @@ -186,3 +186,38 @@ universe CSVs are per-service: persistent on the web disk, ephemeral on the cron The operator runbook (first deploy, secret file, seeding universes) is in [operations.md → "Deploying to Render"](../../operations.md#deploying-to-render-managed). + +## Application storage smoke in the final image + +The quality workflow provisions a PostgreSQL service matching the Compose image for each Python matrix +worker and requires the PostgreSQL fixture lane. The Docker job still verifies +image build and Compose health, then runs: + +```text +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 +``` + +`SCANNER_TEST_POSTGRES_ADMIN_URL` must already be loaded from a protected +**test-server** environment; CI supplies only its disposable Compose template +credentials. Do not point it at a production server. The helper provisions +unique databases, leaving the configured application database untouched. No +market data, AI provider or alert delivery is invoked. The image needs only +runtime dependencies; it does not install pytest/dev tools for the smoke. + +Beginner note: Streamlit's healthy HTTP endpoint proves that its web process is +alive, not that migrations or repository writes work. This command upgrades +seeded historical data, compares migration/ORM schemas and exercises repository +transactions, uniqueness and database cascade inside the actual image. It also +reads packaged curated universe CSVs with an empty relocated DATA_DIR, protecting +the separation of immutable PR116 source lists from generated universe files. +The launcher starts a fresh Python child with DATA_DIR set before imports so +module-level path constants cannot keep the parent's original /data value. +The private directory is removed after the child exits. The PostgreSQL contexts +remove their databases on ordinary success or exception; CI's always-run Compose +teardown removes the entire server/volumes even if a process is forcibly killed. + +A zero exit prints a JSON proof receipt. Any missing opt-in, source packaging +failure, migration error, schema drift or repository violation fails the step. +Database URLs/driver exception text are not printed. Local machines without +Docker or an explicitly configured disposable PostgreSQL server cannot claim +these lanes passed; hosted CI evidence is required before release. diff --git a/docs/architecture/components/storage-persistence.md b/docs/architecture/components/storage-persistence.md index 7d76db8..b7426e6 100644 --- a/docs/architecture/components/storage-persistence.md +++ b/docs/architecture/components/storage-persistence.md @@ -246,3 +246,53 @@ per-signal schema. The PROV-003 ledger accepts richer hashed evidence through new durable identity or query key requires a real schema review. Any schema change requires an Alembic migration, matching `models.py` metadata, repository helpers, and updates to the owning scan-history or IPO design document. + +## Migration verification and disposable database contract + +The read-only comparator in `backend/storage/schema_verification.py` compares +migration-built and ORM-built databases on the **same dialect**. It retains +column types, nullability, server defaults, generated/identity definitions, +ordered primary-key columns, checks, unique constraints, index definitions +(including ordered columns and predicates), and foreign-key targets/options. +`alembic_version` is bookkeeping and excluded; the existing hardcoded table +inventory and exactly-one-Alembic-head guard remain separate checks. + +Beginner note: an index name can survive while its columns or WHERE condition +change. The drift tests deliberately hold names/types constant and change one +real enforcement rule. String-literal whitespace and case stay significant. +Unsupported reflection (for example SQLite expression indexes) fails rather +than silently omitting that index from the evidence. +The only default normalization is SQLite BOOLEAN `'0'`/`'1'` versus `0`/`1`, +whose numeric-affinity equivalence is proved by inserting and reading values. +Foreign-key and primary-key names are not enforcement semantics: SQLite batch +migrations name some FKs that ORM metadata leaves unnamed. Ordered columns, +referred schema/table/columns and every reflected option are still compared. +SQLite PRAGMA also captures actual FK update/delete actions because reflection +of inline REFERENCES syntax can omit those options. + +`tests/conftest.py` owns opt-in PostgreSQL fixtures. Set +`SCANNER_TEST_POSTGRES_ADMIN_URL` to a **test server's** maintenance database URL +(using a role with CREATEDB), stored in a protected environment file rather +than shell arguments. The fixture creates two `scanner_verify_` databases +through the existing pinned psycopg driver. It never clears a shared public +schema or uses the application's DATABASE_URL as a fallback. Both databases +are dropped in finally blocks; a failed second creation also cleans up the +first. Connection/provisioning failures are failures once the URL is supplied. +No URL means a local skip, unless `SCANNER_REQUIRE_POSTGRES_TESTS=1` requires the +lane; CI sets both variables so it cannot silently skip. + +Run the focused contract with: + +```text +python -m pytest tests/test_schema_verification.py tests/test_database_verification_smoke.py tests/test_scan_storage_migrations.py -q +``` + +The shared smoke scenario upgrades the initial scan-history revision, inserts +historical rows using only its original columns, upgrades to the current head, +and checks that IDs, lifecycle status, Decimal price and JSON evidence survive. +It then uses current repository helpers for write/read/delete, checks a real +database uniqueness violation and rollback, and issues a parent DELETE that +bypasses ORM cascade so missing database cascade cannot be masked. Unrelated +historical rows must remain. A second database built from ORM metadata provides +the comparison reference. Existing nonempty verification targets are refused +before DDL. This is a verification tool, not a runtime schema repair routine. diff --git a/tests/conftest.py b/tests/conftest.py index 3fa54b8..58c2787 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -105,3 +105,45 @@ def file_db_engine(tmp_path) -> Iterator[Engine]: def file_session_factory(file_db_engine: Engine) -> SessionFactory: """A transactional session factory bound to the file-backed engine.""" return _session_scope_factory(file_db_engine) + + +@pytest.fixture +def postgres_admin_url() -> str: + """Require opt-in to a PostgreSQL test server, never an application database. + + Returns: + Explicit provisioning URL for a role with CREATEDB permission. + + Beginner note: + Absence is an honest local skip. Once configured, every connection error + fails the tests. CI sets this variable and the separate required flag so + an accidentally removed URL cannot turn its PostgreSQL lane green. + """ + import os + + value = os.environ.get("SCANNER_TEST_POSTGRES_ADMIN_URL", "").strip() + if not value: + if os.environ.get("SCANNER_REQUIRE_POSTGRES_TESTS") == "1": + pytest.fail("SCANNER_TEST_POSTGRES_ADMIN_URL is required for this test lane") + pytest.skip("No disposable PostgreSQL test server configured") + return value + + +@pytest.fixture +def postgres_database_pair(postgres_admin_url: str) -> Iterator[tuple[str, str]]: + """Own two private databases for migrated-versus-ORM schema comparison. + + Yields: + URLs for the migration database and the ORM reference database. + + Beginner note: + Nested cleanup drops the first database even if provisioning the second + fails. Neither fixture rebuilds a shared schema or uses DATABASE_URL. + """ + from backend.storage.verification_database import disposable_postgres_database + + with ( + disposable_postgres_database(postgres_admin_url) as migrated, + disposable_postgres_database(postgres_admin_url) as metadata, + ): + yield migrated, metadata diff --git a/tests/test_database_verification_smoke.py b/tests/test_database_verification_smoke.py new file mode 100644 index 0000000..818881a --- /dev/null +++ b/tests/test_database_verification_smoke.py @@ -0,0 +1,200 @@ +"""Safety and behavior regressions for disposable database verification.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +import pytest +from sqlalchemy import inspect + + +def test_repository_smoke_upgrades_seeded_history(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Execute migrations and the repository scenario on private SQLite files. + + Beginner note: + Empty-database migration success does not prove historical rows survive + an upgrade, nor that the current repository can use the migrated schema. + """ + from backend.storage.verification_smoke import verify_database_pair + + monkeypatch.setenv("DATABASE_URL", "sqlite://") + result = verify_database_pair( + f"sqlite:///{tmp_path / 'migrated.db'}", f"sqlite:///{tmp_path / 'metadata.db'}" + ) + assert result == {"history_preserved": True, "repository_round_trip": True, "schema_converged": True} + + +def test_disposable_postgres_rejects_other_dialects() -> None: + """A test-server setting cannot accidentally authorize SQLite/application DDL. + + Beginner note: + Only a specifically configured PostgreSQL provisioning connection may + create a scratch database; no application URL is an implicit fallback. + """ + from backend.storage.verification_database import disposable_postgres_database + + with pytest.raises(ValueError, match="PostgreSQL"), disposable_postgres_database("sqlite://"): + pytest.fail("Unsupported provisioning URL was accepted") + + +def test_explicit_unreachable_postgres_is_failure() -> None: + """A requested but unreachable server must fail rather than silently skip. + + Beginner note: + This loopback closed-port URL never reaches a provider or application + database. It distinguishes a missing optional setting from broken CI. + """ + from backend.storage.verification_database import disposable_postgres_database + + with ( + pytest.raises(RuntimeError, match="provision disposable PostgreSQL"), + disposable_postgres_database("postgresql+psycopg://test:test@127.0.0.1:1/postgres?connect_timeout=1"), + ): + pytest.fail("The intentionally unreachable server accepted a connection") + + +def test_postgres_migrations_and_repository(postgres_database_pair: tuple[str, str]) -> None: + """Run the same history and enforcement contract against a real PG server. + + Beginner note: + The fixture provisions TWO unique databases, never drops public tables + in a configured database, and fails if an explicitly requested server + is unavailable. CI supplies that setting, so this test cannot skip there. + """ + from backend.storage.verification_smoke import verify_database_pair + + assert verify_database_pair(*postgres_database_pair)["schema_converged"] is True + + +def test_disposable_postgres_cleanup(postgres_admin_url: str) -> None: + """Exception cleanup removes only the database created by this context. + + Beginner note: + The nested fixture outlives the scratch database so the readback can + prove the database was dropped even when the body raises an exception. + """ + from sqlalchemy import create_engine, text + from sqlalchemy.engine import make_url + + from backend.storage.verification_database import disposable_postgres_database + + database_name = "" + with ( + pytest.raises(LookupError, match="intentional fixture failure"), + disposable_postgres_database(postgres_admin_url) as url, + ): + database_name = make_url(url).database or "" + engine = create_engine(url) + try: + assert inspect(engine).get_table_names() == [] + finally: + engine.dispose() + raise LookupError("intentional fixture failure") + engine = create_engine(make_url(postgres_admin_url).set(drivername="postgresql+psycopg")) + try: + with engine.connect() as connection: + assert connection.scalar( + text("SELECT count(*) FROM pg_database WHERE datname = :name"), {"name": database_name} + ) == 0 + finally: + engine.dispose() + + +def test_smoke_cli_requires_explicit_provisioning_url(monkeypatch: pytest.MonkeyPatch) -> None: + """The image CLI must not borrow DATABASE_URL if its test setting is absent. + + Beginner note: + Operators may run a smoke command from an application container. Merely + having production credentials in that container cannot authorize writes. + """ + import os + + env = os.environ.copy() + env.pop("SCANNER_TEST_POSTGRES_ADMIN_URL", None) + env["DATABASE_URL"] = "sqlite:///DO_NOT_CREATE_SMOKE.db" + completed = subprocess.run( + [sys.executable, "-m", "backend.storage.verification_smoke"], + env=env, capture_output=True, text=True, check=False, timeout=30, + ) + assert completed.returncode != 0 + assert "SCANNER_TEST_POSTGRES_ADMIN_URL is required" in completed.stderr + assert not Path("DO_NOT_CREATE_SMOKE.db").exists() + + +def test_required_postgres_lane_cannot_silently_skip() -> None: + """Removing the CI URL must be a failing fixture, not a green skipped lane. + + Beginner note: + This runs only the single PostgreSQL case in a child with no URL, so it + cannot contact a server. The child must fail before provisioning begins. + """ + import os + + env = os.environ.copy() + env.pop("SCANNER_TEST_POSTGRES_ADMIN_URL", None) + env.update(DATABASE_URL="sqlite://", SCANNER_REQUIRE_POSTGRES_TESTS="1") + result = subprocess.run( + [sys.executable, "-m", "pytest", "tests/test_database_verification_smoke.py", + "-k", "test_postgres_migrations_and_repository", "-q", "--tb=short"], + env=env, capture_output=True, text=True, check=False, timeout=60, + ) + assert result.returncode == 1 + assert "SCANNER_TEST_POSTGRES_ADMIN_URL is required for this test lane" in result.stdout + assert "1 skipped" not in result.stdout + + +def test_nonempty_database_is_refused_before_migration(tmp_path: Path) -> None: + """An accidentally reused target must retain its existing table and rows. + + Beginner note: + Provisioning contexts normally guarantee emptiness. This independent + guard ensures direct helper callers cannot rewrite an existing schema. + """ + from sqlalchemy import create_engine, text + + from backend.storage.verification_smoke import verify_database_pair + + url = f"sqlite:///{tmp_path / 'occupied.db'}" + engine = create_engine(url) + try: + with engine.begin() as connection: + connection.execute(text("CREATE TABLE sentinel (marker TEXT)")) + connection.execute(text("INSERT INTO sentinel VALUES ('retain')")) + with pytest.raises(AssertionError, match="must be empty"): + verify_database_pair(url, f"sqlite:///{tmp_path / 'reference.db'}") + with engine.connect() as connection: + assert connection.scalar(text("SELECT marker FROM sentinel")) == "retain" + assert inspect(engine).get_table_names() == ["sentinel"] + finally: + engine.dispose() + + +@pytest.mark.parametrize("source_in_data_dir", [False, True]) +def test_image_sources_are_read_only_and_outside_runtime_data( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, source_in_data_dir: bool +) -> None: + """Read a packaged CSV and reject a source accidentally moved into DATA_DIR. + + Beginner note: + This fixture uses a real CSV parser without fetching an instrument + master. Distinct source/runtime roots are the important image contract; + the current product source-path mapping is checked in hosted smoke. + """ + from backend import config, universe_builder + from backend.storage.verification_smoke import verify_relocated_universe_sources + + data_dir = tmp_path / "runtime" + data_dir.mkdir() + source = (data_dir if source_in_data_dir else tmp_path) / "curated.csv" + content = "symbol\nEXAMPLE\n" + source.write_text(content, encoding="utf-8") + monkeypatch.setattr(config, "DATA_DIR", data_dir) + monkeypatch.setattr(universe_builder, "HEMANT_SOURCE_FILES", {"example": source}) + if source_in_data_dir: + with pytest.raises(AssertionError, match="relocated DATA_DIR"): + verify_relocated_universe_sources() + else: + verify_relocated_universe_sources() + assert source.read_text(encoding="utf-8") == content diff --git a/tests/test_scan_storage_migrations.py b/tests/test_scan_storage_migrations.py index 6031401..9c5f49f 100644 --- a/tests/test_scan_storage_migrations.py +++ b/tests/test_scan_storage_migrations.py @@ -916,31 +916,15 @@ def test_ensure_database_schema_detects_schema_drift(monkeypatch, tmp_path: Path def _reflect_schema(engine) -> dict[str, dict[str, object]]: - """Return a comparable {table: columns/indexes/foreign_keys} structure. + """Reflect full enforcement contracts and dispose this test-owned engine. - Alembic's bookkeeping ``alembic_version`` table is skipped so a migrated - database and a ``create_all`` database describe the same application schema. - The engine is disposed before returning. + Beginner note: + The reusable fingerprint also powers PostgreSQL and image verification; + the wrapper retains this module's existing engine ownership convention. """ - 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"])) - for column in inspector.get_columns(table) - } - indexes = {index["name"] for index in inspector.get_indexes(table)} - foreign_keys = { - ( - tuple(fk["constrained_columns"]), - fk["referred_table"], - tuple(fk["referred_columns"]), - fk.get("options", {}).get("ondelete"), - ) - for fk in inspector.get_foreign_keys(table) - } - schema[table] = {"columns": columns, "indexes": indexes, "foreign_keys": foreign_keys} - engine.dispose() - return schema + from backend.storage.schema_verification import reflect_schema + + try: + return reflect_schema(engine) + finally: + engine.dispose() diff --git a/tests/test_schema_verification.py b/tests/test_schema_verification.py new file mode 100644 index 0000000..a9347c1 --- /dev/null +++ b/tests/test_schema_verification.py @@ -0,0 +1,139 @@ +"""Regression proofs that the schema fingerprint detects enforcement drift.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from sqlalchemy import create_engine +from test_scan_storage_migrations import _reflect_schema + + +@pytest.mark.parametrize( + ("before", "after"), + [ + ("CHECK (amount > 0)", "CHECK (amount >= 0)"), + ("UNIQUE (code)", "UNIQUE (amount)"), + ("PRIMARY KEY (id, code)", "PRIMARY KEY (code, id)"), + ], + ids=["check-expression", "unique-columns", "primary-key-order"], +) +def test_constraint_drift_is_visible(tmp_path: Path, before: str, after: str) -> None: + """Keep identical column types/names while changing a real constraint. + + Beginner note: + The old comparator only saw columns and index names, so it could approve + a migration that accepts forbidden values or changes row identity. + """ + schemas = [] + for position, constraint in enumerate((before, after)): + engine = create_engine(f"sqlite:///{tmp_path / f'{position}.db'}") + with engine.begin() as connection: + connection.exec_driver_sql( + f"CREATE TABLE sample (id INTEGER NOT NULL, code TEXT NOT NULL, " + f"amount INTEGER NOT NULL, {constraint})" + ) + schemas.append(_reflect_schema(engine)) + assert schemas[0] != schemas[1] + + +@pytest.mark.parametrize( + ("before", "after"), + [ + ("CREATE INDEX sample_idx ON sample (id, code)", "CREATE INDEX sample_idx ON sample (code, id)"), + ("CREATE INDEX sample_idx ON sample (code)", "CREATE UNIQUE INDEX sample_idx ON sample (code)"), + ("CREATE INDEX sample_idx ON sample (id) WHERE id > 0", "CREATE INDEX sample_idx ON sample (id) WHERE id >= 0"), + ], + ids=["index-column-order", "index-uniqueness", "index-predicate"], +) +def test_same_named_index_drift_is_visible(tmp_path: Path, before: str, after: str) -> None: + """Changing an index definition must matter even when its name survives. + + Beginner note: + A partial unique index only protects rows matching its predicate. A name + comparison cannot prove either the uniqueness or the protected rows. + """ + schemas = [] + for position, statement in enumerate((before, after)): + engine = create_engine(f"sqlite:///{tmp_path / f'{position}.db'}") + with engine.begin() as connection: + connection.exec_driver_sql("CREATE TABLE sample (id INTEGER, code TEXT)") + connection.exec_driver_sql(statement) + schemas.append(_reflect_schema(engine)) + assert schemas[0] != schemas[1] + + +@pytest.mark.parametrize("defaults", [("0", "1"), ("'A B'", "'A B'"), ("'Mixed'", "'mixed'")]) +def test_server_default_drift_preserves_literal_semantics(tmp_path: Path, defaults: tuple[str, str]) -> None: + """Defaults and significant text inside SQL quotes must stay distinguishable. + + Beginner note: + Removing all whitespace or lowercasing SQL also changes string literals. + These pairs expose normalization that would hide different stored values. + """ + schemas = [] + for position, default in enumerate(defaults): + engine = create_engine(f"sqlite:///{tmp_path / f'{position}.db'}") + with engine.begin() as connection: + connection.exec_driver_sql(f"CREATE TABLE sample (id INTEGER, code TEXT DEFAULT {default})") + schemas.append(_reflect_schema(engine)) + assert schemas[0] != schemas[1] + + +def test_foreign_key_update_action_drift_is_visible(tmp_path: Path) -> None: + """Keep FK names/targets/deletion rules equal while changing update behavior. + + Beginner note: + The previous comparator retained ON DELETE but discarded ON UPDATE; + that omission could approve a migration that breaks parent-key updates. + """ + schemas = [] + for position, action in enumerate(("CASCADE", "RESTRICT")): + engine = create_engine(f"sqlite:///{tmp_path / f'{position}.db'}") + with engine.begin() as connection: + connection.exec_driver_sql("CREATE TABLE parent (id INTEGER PRIMARY KEY)") + connection.exec_driver_sql( + "CREATE TABLE sample (parent_id INTEGER REFERENCES parent(id) " + f"ON DELETE CASCADE ON UPDATE {action})" + ) + schemas.append(_reflect_schema(engine)) + assert schemas[0] != schemas[1] + + +@pytest.mark.parametrize("boolean", [0, 1]) +def test_sqlite_boolean_default_spellings_store_identical_values(tmp_path: Path, boolean: int) -> None: + """Prove the precise SQLite Boolean formatting exception before normalizing. + + Beginner note: + The migration uses DEFAULT 0; SQLAlchemy's string default emits '0'. + SQLite BOOLEAN has numeric affinity, so both store integer zero. This + exception must never be applied to TEXT, where quoted content matters. + """ + schemas = [] + values = [] + for position, default in enumerate((str(boolean), f"'{boolean}'")): + engine = create_engine(f"sqlite:///{tmp_path / f'bool-{position}.db'}") + with engine.begin() as connection: + connection.exec_driver_sql(f"CREATE TABLE sample (id INTEGER, flag BOOLEAN DEFAULT {default})") + connection.exec_driver_sql("INSERT INTO sample (id) VALUES (1)") + values.append(connection.exec_driver_sql("SELECT flag, typeof(flag) FROM sample").one()) + schemas.append(_reflect_schema(engine)) + assert values == [(boolean, "integer"), (boolean, "integer")] + assert schemas[0] == schemas[1] + + +def test_unreflectable_sqlite_expression_index_is_not_silently_omitted(tmp_path: Path) -> None: + """Unsupported reflection must fail closed instead of comparing incomplete schemas. + + Beginner note: + SQLAlchemy warns and skips SQLite expression indexes. Ignoring that + warning would let a missing enforcement index evade this comparator. + """ + from sqlalchemy.exc import SAWarning + + engine = create_engine(f"sqlite:///{tmp_path / 'expression.db'}") + with engine.begin() as connection: + connection.exec_driver_sql("CREATE TABLE sample (code TEXT)") + connection.exec_driver_sql("CREATE UNIQUE INDEX ix_code ON sample (lower(code))") + with pytest.raises(SAWarning, match="expression-based index"): + _reflect_schema(engine) diff --git a/tests/test_supply_chain_policy.py b/tests/test_supply_chain_policy.py index bdc545a..e118b04 100644 --- a/tests/test_supply_chain_policy.py +++ b/tests/test_supply_chain_policy.py @@ -378,3 +378,29 @@ def test_screener_guide_matches_registry_chart_golden_and_ci_contracts(): assert "series.isna().any()" not in text for command in CI_COMMANDS: assert command in text + + +def test_ci_requires_real_postgres_and_application_image_smoke(): + """Protect the executable PostgreSQL and final-image verification lanes. + + Beginner note: + A health endpoint and pip installation cannot prove migrations or + repository writes work in the deployment image. Required opt-in prevents + missing CI environment from silently skipping PostgreSQL integration. + """ + workflow = yaml.safe_load((ROOT / ".github/workflows/quality-and-security.yml").read_text(encoding="utf-8")) + verify = workflow["jobs"]["verify"] + compose = yaml.safe_load((ROOT / "docker-compose.yml").read_text(encoding="utf-8")) + assert verify["services"]["postgres"]["image"] == compose["services"]["postgres"]["image"] + assert verify["env"]["SCANNER_REQUIRE_POSTGRES_TESTS"] == "1" + assert "127.0.0.1:5432/postgres" in verify["env"]["SCANNER_TEST_POSTGRES_ADMIN_URL"] + smoke_steps = workflow["jobs"]["docker-build"]["steps"] + application_smoke = next( + step for step in smoke_steps if step["name"] == "Verify application storage in final image" + ) + assert "docker compose run --rm -T --no-deps" in application_smoke["run"] + assert "-e SCANNER_TEST_POSTGRES_ADMIN_URL" in application_smoke["run"] + assert "scanner-ui python -m backend.storage.verification_smoke" in application_smoke["run"] + assert "@postgres:5432/postgres" in application_smoke["env"]["SCANNER_TEST_POSTGRES_ADMIN_URL"] + teardown = next(step for step in smoke_steps if step["name"] == "Tear down Compose stack") + assert teardown["if"] == "always()" From ce670ec6027265ecd86c7bf01702d1bfb4e7bea6 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Sun, 4 Oct 2026 22:46:48 +0530 Subject: [PATCH 2/5] fix(verification): contain setup errors and clean up committed test databases Keep URL and engine configuration failures secret-safe; record successful CREATE before connection cleanup can fail. Co-authored-by: Codex --- backend/storage/verification_database.py | 29 +++++--- tests/test_database_verification_smoke.py | 81 +++++++++++++++++++++++ 2 files changed, 102 insertions(+), 8 deletions(-) diff --git a/backend/storage/verification_database.py b/backend/storage/verification_database.py index e90e17f..a6147e8 100644 --- a/backend/storage/verification_database.py +++ b/backend/storage/verification_database.py @@ -8,6 +8,7 @@ from sqlalchemy import create_engine from sqlalchemy.engine import make_url +from sqlalchemy.exc import ArgumentError from sqlalchemy.pool import NullPool @@ -23,8 +24,9 @@ def disposable_postgres_database(admin_url: str) -> Iterator[str]: A URL for the newly created private database, using the pinned driver. Raises: - ValueError: The URL is not PostgreSQL. - RuntimeError: Provisioning or cleanup fails; credentials are suppressed. + ValueError: The URL cannot be parsed or does not name PostgreSQL. + 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 @@ -33,23 +35,34 @@ def disposable_postgres_database(admin_url: str) -> Iterator[str]: cannot run in a transaction, and pooled scratch connections block DROP. FORCE terminates only connections to our own scratch database on supported PostgreSQL. """ - url = make_url(admin_url) + # 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") 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 - engine = create_engine( - url, isolation_level="AUTOCOMMIT", poolclass=NullPool, - connect_args={"connect_timeout": 5}, - ) + 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') - created = True + # 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" diff --git a/tests/test_database_verification_smoke.py b/tests/test_database_verification_smoke.py index 818881a..8517dbe 100644 --- a/tests/test_database_verification_smoke.py +++ b/tests/test_database_verification_smoke.py @@ -39,6 +39,87 @@ def test_disposable_postgres_rejects_other_dialects() -> None: pytest.fail("Unsupported provisioning URL was accepted") +def test_disposable_postgres_hides_malformed_url_components() -> None: + """A malformed provisioning URL must not echo its opaque contents. + + Beginner note: + URL parsing used to happen before the safe-error boundary. A malformed + port was repeated verbatim by ValueError even though the helper promises + secret-safe failures. This fails before any connection can be attempted. + """ + from backend.storage.verification_database import disposable_postgres_database + + marker = "PRIVATE_CONFIGURATION_VALUE" + with pytest.raises(ValueError) as error, disposable_postgres_database( + f"postgresql://fixture:fixture@127.0.0.1:{marker}/postgres" + ): + pytest.fail("Malformed connection URL was accepted") + assert marker not in str(error.value) + assert "valid PostgreSQL URL" in str(error.value) + + +def test_disposable_postgres_hides_engine_configuration_errors() -> None: + """Engine construction must use the same safe boundary as connection setup. + + Beginner note: + An unknown SQLAlchemy plugin fails before a connection exists and echoes + its URL option. This fake option demonstrates the leak without reaching + a server; all configuration details must stay out of the public error. + """ + from backend.storage.verification_database import disposable_postgres_database + + marker = "PRIVATE_UNKNOWN_PLUGIN" + with pytest.raises(Exception) as error, disposable_postgres_database( + f"postgresql://fixture:fixture@127.0.0.1/postgres?plugin={marker}" + ): + pytest.fail("An unknown engine plugin was accepted") + assert marker not in str(error.value) + assert isinstance(error.value, RuntimeError) + + +def test_disposable_postgres_cleans_up_when_create_connection_exit_fails(monkeypatch: pytest.MonkeyPatch) -> None: + """Track a successful CREATE before its connection context can raise. + + Beginner note: + Creation is committed immediately in AUTOCOMMIT. If connection cleanup + then fails, the private database still exists and must be dropped. This + fixture records DDL without reaching a server and proves the DROP target + is exactly the generated CREATE target, never the configured database. + """ + from contextlib import contextmanager + from types import SimpleNamespace + + from backend.storage import verification_database + + commands: list[str] = [] + disposed: list[bool] = [] + + @contextmanager + def connection(): + """Model a close failure only after the first committed DDL statement. + + Beginner note: + The second context is the finally cleanup path and closes normally; + changing the flag after the first context exits would miss that path. + """ + yield SimpleNamespace(exec_driver_sql=commands.append) + if len(commands) == 1: + raise RuntimeError("Fixture close failure") + + monkeypatch.setattr(verification_database, "create_engine", lambda *_args, **_kwargs: SimpleNamespace( + connect=connection, dispose=lambda: disposed.append(True), + )) + with pytest.raises(RuntimeError, match="Cannot provision"), verification_database.disposable_postgres_database( + "postgresql://fixture:fixture@127.0.0.1/existing_database" + ): + pytest.fail("A failed provisioning connection was yielded") + assert len(commands) == 2 + created_name = commands[0].split('"')[1] + assert created_name.startswith("scanner_verify_") + assert commands[1] == f'DROP DATABASE "{created_name}" WITH (FORCE)' + assert disposed == [True] + + def test_explicit_unreachable_postgres_is_failure() -> None: """A requested but unreachable server must fail rather than silently skip. From 7dc1ade8474103e3ca7d6ced78ee3f7aa1d9025f Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Mon, 5 Oct 2026 22:25:11 +0530 Subject: [PATCH 3/5] fix(types): annotate historical smoke-test database identities Co-authored-by: Codex --- backend/storage/verification_smoke.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/storage/verification_smoke.py b/backend/storage/verification_smoke.py index 6aa5852..072363a 100644 --- a/backend/storage/verification_smoke.py +++ b/backend/storage/verification_smoke.py @@ -72,11 +72,11 @@ def _seed_history(engine: Engine) -> tuple[int, int]: lets both dialects generate IDs so sequence state remains realistic. """ with engine.begin() as connection: - run_id = connection.execute(text( + run_id: int = connection.execute(text( "INSERT INTO scan_runs (started_at, finished_at, status, screener_key, universe_key) " "VALUES (:started, :started, 'success', 'verification_legacy', 'nifty_100') RETURNING id" ), {"started": dt.datetime(2026, 6, 4, tzinfo=dt.UTC).isoformat()}).scalar_one() - result_id = connection.execute(text( + result_id: int = connection.execute(text( "INSERT INTO scan_results (run_id, symbol, close_price, raw_result_json, created_at) " "VALUES (:run_id, 'LEGACY', 123.4500, :payload, :created) RETURNING id" ), {"run_id": run_id, "payload": '{"symbol":"LEGACY","history_marker":"retain"}', From c035b7f55cb5623bb6cdc3b56cdfa50abae07923 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Sun, 11 Oct 2026 11:56:58 +0530 Subject: [PATCH 4/5] fix: preserve schema semantics and disposable database isolation Compare SQLite index collations and ordering, including implicit constraint indexes. Reject PostgreSQL query database overrides before provisioning. Add real PostgreSQL IPO concurrency and rollback scenarios, and document helper contracts and verification limits. Co-authored-by: Codex --- backend/storage/schema_verification.py | 48 ++++ backend/storage/verification_database.py | 12 +- backend/storage/verification_smoke.py | 49 ++++ .../components/storage-persistence.md | 22 +- tests/conftest.py | 39 +++ tests/test_database_verification_smoke.py | 66 +++++ tests/test_ipo_postgres_concurrency.py | 239 ++++++++++++++++++ tests/test_schema_verification.py | 84 ++++++ 8 files changed, 555 insertions(+), 4 deletions(-) create mode 100644 tests/test_ipo_postgres_concurrency.py diff --git a/backend/storage/schema_verification.py b/backend/storage/schema_verification.py index a6053d5..f42f6d1 100644 --- a/backend/storage/schema_verification.py +++ b/backend/storage/schema_verification.py @@ -40,6 +40,44 @@ def _freeze(value: Any) -> Any: 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. @@ -50,6 +88,13 @@ def reflect_schema(engine: Engine) -> dict[str, dict[str, object]]: 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: @@ -109,6 +154,9 @@ def reflect_schema(engine: Engine) -> dict[str, dict[str, object]]: "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, } diff --git a/backend/storage/verification_database.py b/backend/storage/verification_database.py index a6147e8..c0c93f7 100644 --- a/backend/storage/verification_database.py +++ b/backend/storage/verification_database.py @@ -18,13 +18,15 @@ def disposable_postgres_database(admin_url: str) -> Iterator[str]: Args: admin_url: Explicit test-server URL whose role can CREATE DATABASE. - Its configured database is only a maintenance connection target. + 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 or does not name PostgreSQL. + 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. @@ -44,6 +46,12 @@ def disposable_postgres_database(admin_url: str) -> Iterator[str]: 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. diff --git a/backend/storage/verification_smoke.py b/backend/storage/verification_smoke.py index 072363a..724131e 100644 --- a/backend/storage/verification_smoke.py +++ b/backend/storage/verification_smoke.py @@ -35,6 +35,12 @@ def _migration_url(url: str) -> Iterator[None]: """Temporarily direct Alembic's environment reader at a private database. + Args: + url: Caller-provisioned scratch database URL for this synchronous block. + + Yields: + None while ``DATABASE_URL`` names the scratch database. + Beginner note: Alembic reads DATABASE_URL at invocation time. Restore the original value even on failure so an integration test cannot redirect later tests. @@ -54,6 +60,10 @@ def _migration_url(url: str) -> Iterator[None]: def migration_config() -> Config: """Return a config anchored to the installed application migration directory. + Returns: + An Alembic configuration that locates the packaged migration scripts. + The database URL is supplied separately by ``_migration_url``. + Beginner note: A programmatic Config preserves the application's logging/redaction setup. Resolving from __file__ also works from a non-repository cwd. @@ -66,6 +76,16 @@ def migration_config() -> Config: def _seed_history(engine: Engine) -> tuple[int, int]: """Insert initial-revision rows using only columns that existed then. + Args: + engine: Caller-owned scratch engine upgraded to ``20260604scan002``. + + Returns: + The committed historical scan-run and result IDs, in that order. + + Raises: + sqlalchemy.exc.SQLAlchemyError: The historical insert cannot complete; + the engine transaction rolls back both fixture rows together. + Beginner note: Current ORM INSERTs cannot seed an old schema: they mention columns which do not exist yet. Bound SQL uses the real historical shape and @@ -87,6 +107,16 @@ def _seed_history(engine: Engine) -> tuple[int, int]: def _verify_repository(engine: Engine, legacy_run_id: int, legacy_result_id: int) -> None: """Prove durable repository writes, rollback, uniqueness and database cascade. + Args: + engine: Caller-owned private engine already migrated to the current head. + legacy_run_id: Historical run identity returned by ``_seed_history``. + legacy_result_id: Historical result identity returned by that helper. + + Raises: + AssertionError: A persisted value or database-enforced invariant differs + from the expected history, transaction, uniqueness, or cascade behavior. + sqlalchemy.exc.SQLAlchemyError: An unexpected database operation fails. + Beginner note: Separate transactions ensure reads observe committed storage rather than a Session identity map. A Core parent DELETE bypasses ORM cascades, so @@ -160,6 +190,9 @@ def verify_database_pair(migrated_url: str, metadata_url: str) -> dict[str, bool Raises: AssertionError: Database identity, history, repository, or schema differs. + alembic.util.exc.CommandError: The migration chain cannot be loaded or applied. + sqlalchemy.exc.SQLAlchemyError: Database access or schema reflection fails. + sqlalchemy.exc.SAWarning: Reflection cannot describe an index completely. Beginner note: Refuse nonempty databases BEFORE any DDL. Provisioning and eventual @@ -204,6 +237,10 @@ def verify_database_pair(migrated_url: str, metadata_url: str) -> dict[str, bool def verify_relocated_universe_sources() -> None: """Read packaged source universes after moving mutable data to an empty dir. + Raises: + AssertionError: A curated source follows mutable ``DATA_DIR`` or does + not yield any symbols through the application's CSV loader. + Beginner note: PR116 separates curated source CSVs from generated DATA_DIR files. This check must read the package's sources even when the private runtime @@ -222,6 +259,11 @@ def verify_relocated_universe_sources() -> None: def run_image_verification() -> None: """Verify image sources/storage after the launcher sets private environment. + Raises: + SystemExit: Verification or scratch-database cleanup fails. Fixed + assertion diagnostics remain readable; unexpected exception messages + are replaced with their type so database credentials cannot leak. + Beginner note: Run in the launcher's fresh process so import-time path constants read the relocated DATA_DIR. Suppress driver errors to avoid echoing a DSN. @@ -245,6 +287,13 @@ def run_image_verification() -> None: def main() -> int: """Run image verification using explicitly authorized scratch resources. + Returns: + Zero after the isolated child completes all source and storage checks. + + Raises: + SystemExit: The explicit test-server setting is absent or the child fails. + subprocess.TimeoutExpired: The child exceeds the 180-second smoke budget. + Beginner note: The package creates its normal lazy engine at import time but does not connect to it. The child replaces DATABASE_URL with harmless SQLite diff --git a/docs/architecture/components/storage-persistence.md b/docs/architecture/components/storage-persistence.md index 20eb2c1..7175e57 100644 --- a/docs/architecture/components/storage-persistence.md +++ b/docs/architecture/components/storage-persistence.md @@ -269,7 +269,12 @@ Foreign-key and primary-key names are not enforcement semantics: SQLite batch migrations name some FKs that ORM metadata leaves unnamed. Ordered columns, referred schema/table/columns and every reflected option are still compared. SQLite PRAGMA also captures actual FK update/delete actions because reflection -of inline REFERENCES syntax can omit those options. +of inline REFERENCES syntax can omit those options. `index_list` and +`index_xinfo` retain collations and sort directions for explicit indexes and +the implicit indexes backing UNIQUE/PRIMARY KEY constraints. Generated index +names are excluded because their suffixes depend on declaration order. +The collation regression inserts case-distinct values: one schema accepts them +and the other rejects them, proving an enforcement difference beyond DDL text. `tests/conftest.py` owns opt-in PostgreSQL fixtures. Set `SCANNER_TEST_POSTGRES_ADMIN_URL` to a **test server's** maintenance database URL @@ -278,7 +283,10 @@ than shell arguments. The fixture creates two `scanner_verify_` databases through the existing pinned psycopg driver. It never clears a shared public schema or uses the application's DATABASE_URL as a fallback. Both databases are dropped in finally blocks; a failed second creation also cleans up the -first. Connection/provisioning failures are failures once the URL is supplied. +first. A query `dbname` option is rejected before engine construction because +the driver would otherwise let it override the generated scratch database. +Put the maintenance database in the URL path; TLS options remain supported. +Connection/provisioning failures are failures once the URL is supplied. No URL means a local skip, unless `SCANNER_REQUIRE_POSTGRES_TESTS=1` requires the lane; CI sets both variables so it cannot silently skip. @@ -297,3 +305,13 @@ bypasses ORM cascade so missing database cascade cannot be masked. Unrelated historical rows must remain. A second database built from ORM metadata provides the comparison reference. Existing nonempty verification targets are refused before DDL. This is a verification tool, not a runtime schema repair routine. + +`tests/test_ipo_postgres_concurrency.py` uses a separate migrated database per +scenario to exercise the IPO state contract with real PostgreSQL sessions: +identical concurrent calculations publish one complete pair, a paused older +calculation cannot supersede newer evidence, and an outer rollback removes a +staged evaluation together with its input revision. Barriers and events establish +the intended ordering; bounded waits and database timeouts limit failed tests. +An independent reader checks committed state while a writer remains open. +SQLite tests of the same scenarios are useful local checks, but they cannot +establish PostgreSQL row-lock behavior. The required hosted lane must run them. diff --git a/tests/conftest.py b/tests/conftest.py index 58c2787..028635f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -147,3 +147,42 @@ def postgres_database_pair(postgres_admin_url: str) -> Iterator[tuple[str, str]] disposable_postgres_database(postgres_admin_url) as metadata, ): yield migrated, metadata + + +@pytest.fixture +def postgres_session_factory(postgres_admin_url: str) -> Iterator[SessionFactory]: + """Yield independent transactions on one private, fully migrated database. + + Args: + postgres_admin_url: Explicit test-server provisioning URL, never DATABASE_URL. + + Yields: + Commit-on-success session scopes using distinct pooled connections. + + Beginner note: + Migrate synchronously before any worker starts: Alembic temporarily reads + DATABASE_URL, which is process-wide. Real PostgreSQL transactions then + exercise row locking rather than SQLite's database-wide writer lock. + Server-side timeouts bound broken locks even if a worker cannot finish; + dispose all pooled connections before dropping the UUID-owned database. + """ + from alembic import command + + from backend.storage.verification_database import disposable_postgres_database + from backend.storage.verification_smoke import _migration_url, migration_config + + with disposable_postgres_database(postgres_admin_url) as database_url: + with _migration_url(database_url): + command.upgrade(migration_config(), "head") + engine = create_engine( + database_url, + future=True, + connect_args={ + "connect_timeout": 5, + "options": "-c statement_timeout=15000 -c lock_timeout=10000", + }, + ) + try: + yield _session_scope_factory(engine) + finally: + engine.dispose() diff --git a/tests/test_database_verification_smoke.py b/tests/test_database_verification_smoke.py index 8517dbe..865a0ab 100644 --- a/tests/test_database_verification_smoke.py +++ b/tests/test_database_verification_smoke.py @@ -77,6 +77,72 @@ def test_disposable_postgres_hides_engine_configuration_errors() -> None: assert isinstance(error.value, RuntimeError) +def test_disposable_postgres_rejects_database_override_before_engine_setup(monkeypatch: pytest.MonkeyPatch) -> None: + """Reject a query database that would override both generated scratch URLs. + + Beginner note: + SQLAlchemy lets a ``dbname`` query option override the URL path passed + to psycopg. Replacing only the path used to leave both supposedly private + URLs targeting the caller's database. Refusal must happen before CREATE + or any connection, and diagnostics must not repeat the supplied value. + """ + from backend.storage import verification_database + + created_engines = [] + + def unexpected_engine(*args, **kwargs): + """Record an unsafe early engine setup without contacting PostgreSQL.""" + created_engines.append(True) + raise AssertionError("Database override reached engine construction") + + monkeypatch.setattr(verification_database, "create_engine", unexpected_engine) + marker = "PRIVATE_EXISTING_DATABASE" + with pytest.raises(ValueError, match="query") as error, verification_database.disposable_postgres_database( + f"postgresql://fixture:fixture@127.0.0.1/postgres?dbname={marker}" + ): + pytest.fail("A database-overriding URL was accepted") + assert created_engines == [] + assert marker not in str(error.value) + + +def test_disposable_postgres_driver_targets_generated_database_and_keeps_tls(monkeypatch: pytest.MonkeyPatch) -> None: + """Check actual driver arguments, not just the displayed scratch URL path. + + Beginner note: + A safe-looking URL is insufficient when query options can override its + path. This no-network fixture runs SQLAlchemy's real psycopg argument + conversion and proves the generated CREATE target is also the effective + connection database. Legitimate TLS options must survive the boundary. + """ + from contextlib import contextmanager + from types import SimpleNamespace + + from sqlalchemy.dialects.postgresql.psycopg import PGDialect_psycopg + from sqlalchemy.engine import make_url + + from backend.storage import verification_database + + commands: list[str] = [] + + @contextmanager + def connection(): + """Record scratch CREATE/DROP statements without opening a connection.""" + yield SimpleNamespace(exec_driver_sql=commands.append) + + monkeypatch.setattr(verification_database, "create_engine", lambda *_args, **_kwargs: SimpleNamespace( + connect=connection, dispose=lambda: None, + )) + with verification_database.disposable_postgres_database( + "postgresql://fixture:fixture@127.0.0.1/postgres?sslmode=verify-full" + ) as private_url: + created_name = commands[0].split('"')[1] + _args, kwargs = PGDialect_psycopg().create_connect_args(make_url(private_url)) + assert kwargs["dbname"] == created_name + assert created_name.startswith("scanner_verify_") + assert kwargs["sslmode"] == "verify-full" + assert commands[1] == f'DROP DATABASE "{created_name}" WITH (FORCE)' + + def test_disposable_postgres_cleans_up_when_create_connection_exit_fails(monkeypatch: pytest.MonkeyPatch) -> None: """Track a successful CREATE before its connection context can raise. diff --git a/tests/test_ipo_postgres_concurrency.py b/tests/test_ipo_postgres_concurrency.py new file mode 100644 index 0000000..29fdc58 --- /dev/null +++ b/tests/test_ipo_postgres_concurrency.py @@ -0,0 +1,239 @@ +"""IPO scoring transaction contracts against disposable migrated PostgreSQL. + +Beginner note: + SQLite's single-writer behavior cannot prove PostgreSQL row-lock semantics. + These tests opt in through SCANNER_TEST_POSTGRES_ADMIN_URL and use separate + real sessions. Synchronization pauses calculation, never replaces persistence. + Missing local PostgreSQL is a skip; the required CI lane treats absence as + failure. No shared application database or external evidence source is used. +""" + +from __future__ import annotations + +import datetime as dt +from collections.abc import Iterator +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from decimal import Decimal +from pathlib import Path +from threading import Barrier, Event + +import pytest +from sqlalchemy import func, select +from sqlalchemy.orm import Session +from test_ipo_scoring_service import _AS_OF, _issue_data, _scored_issue + +from backend.ipo import repository +from backend.ipo.repository import SessionFactory +from backend.ipo.scoring import service +from backend.ipo.scoring.state import IpoScoringSnapshot +from backend.storage.ipo_repository import get_ipo_scoring_state_values, update_ipo_issue_row +from backend.storage.models import IpoRecommendation, IpoScore + + +def _history_counts(session: Session, issue_id: int) -> tuple[int, int]: + """Count both halves of persisted evaluation history independently. + + Args: + session: Caller-owned inspection transaction. + issue_id: Issue whose immutable history is inspected. + + Returns: + Score count and recommendation count, including incomplete pairs. + + Beginner note: + The public history list joins complete pairs. Counting both tables also + catches an orphan insert that a successful joined read could conceal. + """ + scores = session.scalar(select(func.count()).select_from(IpoScore).where(IpoScore.issue_id == issue_id)) + recommendations = session.scalar( + select(func.count()).select_from(IpoRecommendation).where( + IpoRecommendation.score_id.in_(select(IpoScore.id).where(IpoScore.issue_id == issue_id)) + ) + ) + assert scores is not None and recommendations is not None + return scores, recommendations + + +def test_postgres_identical_concurrent_calculations_publish_one_pair( + postgres_session_factory: SessionFactory, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Concurrent identical jobs return the same complete, current evaluation. + + Beginner note: + Both jobs finish their real reads before either publishes. Removing the + publication serialization/idempotency guard would allow duplicate pairs, + leak an integrity failure, or leave current state pointing at other history. + A barrier establishes the overlap; no elapsed-time assertion is involved. + """ + issue = _scored_issue(postgres_session_factory, tmp_path) + snapshots_ready = Barrier(2) + load_snapshot = service.load_ipo_scoring_snapshot + + def synchronized_snapshot( + issue_id: int, *, as_of: dt.datetime, session_factory: SessionFactory = postgres_session_factory + ) -> IpoScoringSnapshot: + """Detach real evidence, then release both calculations together. + + Args: + issue_id: Issue being scored. + as_of: Frozen business clock supplied by the scoring service. + session_factory: Real transaction factory forwarded unchanged. + + Returns: + The unchanged, revision-validated evidence snapshot. + """ + snapshot = load_snapshot(issue_id, as_of=as_of, session_factory=session_factory) + snapshots_ready.wait(timeout=15) + return snapshot + + monkeypatch.setattr(service, "load_ipo_scoring_snapshot", synchronized_snapshot) + with ThreadPoolExecutor(max_workers=2) as workers: + futures = [ + workers.submit(service.rescore_issue, issue.id, as_of=_AS_OF, session_factory=postgres_session_factory) + for _ in range(2) + ] + outcomes = [future.result(timeout=30) for future in futures] + # Restore before the current reader, which also loads a scoring snapshot. + monkeypatch.setattr(service, "load_ipo_scoring_snapshot", load_snapshot) + assert {outcome.status for outcome in outcomes} == {"evaluated", "skipped_unchanged"} + evaluation = outcomes[0].evaluation + assert evaluation is not None and outcomes[1].evaluation == evaluation + with postgres_session_factory() as session: + assert _history_counts(session, issue.id) == (1, 1) + current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory) + assert current.fresh and current.evaluation == evaluation + assert current.snapshot.state.current_score_id == evaluation.score_id + assert current.snapshot.state.evaluated_revision == current.snapshot.state.input_revision + + +def test_postgres_older_calculation_cannot_supersede_newer_revision( + postgres_session_factory: SessionFactory, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An older detached calculation retries and reuses the newer committed pair. + + Beginner note: + Pause after the old snapshot's transaction closes, then commit changed + pricing and its evaluation through independent sessions. If publication + loses its revision compare-and-set, releasing the old calculation would + insert obsolete history or supersede the newer current receipt. + """ + issue = _scored_issue(postgres_session_factory, tmp_path) + old_snapshot_ready = Event() + newer_committed = Event() + load_snapshot = service.load_ipo_scoring_snapshot + + def paused_factory() -> Iterator[Session]: + """Identify the old worker without changing transaction behavior. + + Yields: + A real independent session, committed or rolled back by the fixture. + + Beginner note: + The wrapper's identity limits the pause to the old worker. Its + transactions still come from the shared PostgreSQL engine; no lock + or persistence operation is mocked. + """ + with postgres_session_factory() as session: + yield session + + old_worker_factory = contextmanager(paused_factory) + + def pause_old_snapshot( + issue_id: int, *, as_of: dt.datetime, session_factory: SessionFactory = postgres_session_factory + ) -> IpoScoringSnapshot: + """Hold only the old worker's first detached evidence bundle. + + Args: + issue_id: Issue being scored. + as_of: Service's frozen business time. + session_factory: Identifies the old worker while forwarding real I/O. + + Returns: + An unchanged snapshot after the newer commit, or immediately on retry. + """ + snapshot = load_snapshot(issue_id, as_of=as_of, session_factory=session_factory) + if session_factory is old_worker_factory and not old_snapshot_ready.is_set(): + old_snapshot_ready.set() + assert newer_committed.wait(timeout=20), "Newer revision was never committed" + return snapshot + + monkeypatch.setattr(service, "load_ipo_scoring_snapshot", pause_old_snapshot) + with ThreadPoolExecutor(max_workers=1) as workers: + older = workers.submit(service.rescore_issue, issue.id, as_of=_AS_OF, session_factory=old_worker_factory) + try: + assert old_snapshot_ready.wait(timeout=15), "Old worker did not capture its snapshot" + repository.update_issue( + issue.id, _issue_data(price_band_high=Decimal("300")), session_factory=postgres_session_factory + ) + newer = service.rescore_issue(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory) + assert newer.status == "evaluated" and newer.evaluation is not None + committed = service.get_current_evaluation( + issue.id, as_of=_AS_OF, session_factory=postgres_session_factory + ) + finally: + # Release on assertion/error too, so executor cleanup never hangs. + newer_committed.set() + retried = older.result(timeout=30) + assert retried.status == "skipped_unchanged" and retried.evaluation == newer.evaluation + assert repository.list_evaluations(issue.id, session_factory=postgres_session_factory) == [newer.evaluation] + current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory) + assert current.fresh and current.evaluation == newer.evaluation + assert current.snapshot.state.input_revision == committed.snapshot.state.input_revision + assert current.snapshot.state.evaluated_revision == committed.snapshot.state.input_revision + with postgres_session_factory() as session: + assert _history_counts(session, issue.id) == (1, 1) + + +def test_postgres_caller_rollback_restores_evidence_revision_and_selection( + postgres_session_factory: SessionFactory, tmp_path: Path +) -> None: + """An outer caller failure rolls back input and nested score publication. + + Beginner note: + A service may flush and use a savepoint, but the caller owns commit. + Independent readers must see the original receipt while writes are open, + and after rollback both the evidence token and immutable history must be + restored. This catches an internal commit or escaped savepoint insertion. + """ + issue = _scored_issue(postgres_session_factory, tmp_path) + original = service.rescore_issue(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory) + assert original.evaluation is not None + with postgres_session_factory() as session: + original_state = get_ipo_scoring_state_values(session, issue.id) + assert original_state is not None + + with pytest.raises(RuntimeError, match="caller aborts transaction"), postgres_session_factory() as writer: + changed = update_ipo_issue_row(writer, issue.id, {"price_band_high": Decimal("300")}) + assert changed is not None + + @contextmanager + def caller_session() -> Iterator[Session]: + """Lend the outer session while retaining the caller's commit authority. + + Yields: + The existing caller transaction, without commit or rollback. + + Beginner note: + Snapshot reads and publication share this transaction so the + final caller rollback includes evidence and the nested pair. + """ + yield writer + + staged = service.rescore_issue(issue.id, as_of=_AS_OF, session_factory=caller_session) + assert staged.status == "evaluated" and staged.evaluation is not None + staged_state = get_ipo_scoring_state_values(writer, issue.id) + assert staged_state is not None and staged_state[0] > original_state[0] + assert staged_state[1] == staged.evaluation.score_id and staged_state[2] == staged_state[0] + assert _history_counts(writer, issue.id) == (2, 2) + with postgres_session_factory() as reader: + assert get_ipo_scoring_state_values(reader, issue.id) == original_state + assert _history_counts(reader, issue.id) == (1, 1) + raise RuntimeError("caller aborts transaction") + + with postgres_session_factory() as reader: + assert get_ipo_scoring_state_values(reader, issue.id) == original_state + assert _history_counts(reader, issue.id) == (1, 1) + current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=postgres_session_factory) + assert current.fresh and current.evaluation == original.evaluation + assert current.snapshot.inputs.issue.price_band_high == Decimal("242") diff --git a/tests/test_schema_verification.py b/tests/test_schema_verification.py index a9347c1..df37cd9 100644 --- a/tests/test_schema_verification.py +++ b/tests/test_schema_verification.py @@ -6,6 +6,7 @@ import pytest from sqlalchemy import create_engine +from sqlalchemy.exc import IntegrityError from test_scan_storage_migrations import _reflect_schema @@ -63,6 +64,89 @@ def test_same_named_index_drift_is_visible(tmp_path: Path, before: str, after: s assert schemas[0] != schemas[1] +@pytest.mark.parametrize("index_kind", ["index", "unique", "primary_key"]) +def test_sqlite_unique_index_collation_drift_is_visible(tmp_path: Path, index_kind: str) -> None: + """Expose different accepted values despite identical reflected index columns. + + Beginner note: + SQLAlchemy omits SQLite index collations without a warning. Previously + the comparator treated BINARY and NOCASE unique indexes as equivalent, + even though only one accepts two differently capitalized spellings. + Table-level UNIQUE and PRIMARY KEY rules use implicit indexes and need + the same check. + """ + schemas = [] + accepted_case_variant = [] + for position, collation in enumerate(("BINARY", "NOCASE")): + engine = create_engine(f"sqlite:///{tmp_path / f'collation-{position}.db'}") + try: + with engine.begin() as connection: + if index_kind != "index": + constraint = "UNIQUE" if index_kind == "unique" else "PRIMARY KEY" + connection.exec_driver_sql( + f"CREATE TABLE sample (code TEXT, {constraint} (code COLLATE {collation}))" + ) + else: + connection.exec_driver_sql("CREATE TABLE sample (code TEXT)") + connection.exec_driver_sql( + f"CREATE UNIQUE INDEX sample_idx ON sample (code COLLATE {collation})" + ) + connection.exec_driver_sql("INSERT INTO sample VALUES ('Mixed')") + try: + connection.exec_driver_sql("INSERT INTO sample VALUES ('mixed')") + except IntegrityError: + accepted_case_variant.append(False) + else: + accepted_case_variant.append(True) + schemas.append(_reflect_schema(engine)) + finally: + engine.dispose() + assert accepted_case_variant == [True, False] + assert schemas[0] != schemas[1] + + +def test_sqlite_implicit_index_numbering_does_not_create_false_drift(tmp_path: Path) -> None: + """Compare constraint definitions independently of SQLite autoindex suffixes. + + Beginner note: + Reordering two equivalent UNIQUE declarations changes SQLite's generated + index names. The new collation guard must retain both constraints while + ignoring those incidental names, so harmless migration formatting does + not become a false schema mismatch. + """ + schemas = [] + for position, constraints in enumerate(("UNIQUE (id), UNIQUE (code)", "UNIQUE (code), UNIQUE (id)")): + engine = create_engine(f"sqlite:///{tmp_path / f'autoindex-{position}.db'}") + try: + with engine.begin() as connection: + connection.exec_driver_sql(f"CREATE TABLE sample (id INTEGER, code TEXT, {constraints})") + schemas.append(_reflect_schema(engine)) + finally: + engine.dispose() + assert schemas[0] == schemas[1] + + +def test_sqlite_index_sort_direction_drift_is_visible(tmp_path: Path) -> None: + """Retain descending index order that SQLite's generic reflection drops. + + Beginner note: + Both indexes have the same name and column list, but the second key's + direction changes the ordering that the index supports. The fingerprint + must compare the actual key definition, not just the column names. + """ + schemas = [] + for position, direction in enumerate(("ASC", "DESC")): + engine = create_engine(f"sqlite:///{tmp_path / f'direction-{position}.db'}") + try: + with engine.begin() as connection: + connection.exec_driver_sql("CREATE TABLE sample (id INTEGER, code TEXT)") + connection.exec_driver_sql(f"CREATE INDEX sample_idx ON sample (id ASC, code {direction})") + schemas.append(_reflect_schema(engine)) + finally: + engine.dispose() + assert schemas[0] != schemas[1] + + @pytest.mark.parametrize("defaults", [("0", "1"), ("'A B'", "'A B'"), ("'Mixed'", "'mixed'")]) def test_server_default_drift_preserves_literal_semantics(tmp_path: Path, defaults: tuple[str, str]) -> None: """Defaults and significant text inside SQL quotes must stay distinguishable. From 1b4e38e7b1ccc34ec9c9c57b55a4a51f17ed3068 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Tue, 13 Oct 2026 18:33:33 +0530 Subject: [PATCH 5/5] fix(ci): resolve runner temporary paths at test step scope GitHub evaluates job-level environment before the runner context exists. Keep the private file database on the test step and guard its supported context in the workflow policy regression. Co-authored-by: Codex --- .github/workflows/quality-and-security.yml | 8 +++++--- tests/test_supply_chain_policy.py | 7 ++++++- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/.github/workflows/quality-and-security.yml b/.github/workflows/quality-and-security.yml index ae3c644..558581f 100644 --- a/.github/workflows/quality-and-security.yml +++ b/.github/workflows/quality-and-security.yml @@ -40,9 +40,6 @@ jobs: --health-cmd "pg_isready -U scanner_test -d postgres" --health-interval 5s --health-timeout 5s --health-retries 10 env: - # Alembic and application sessions use separate engines. A runner-owned - # file shares their schema safely without touching configured app data. - DATABASE_URL: sqlite:///${{ runner.temp }}/scanner-default.db SCANNER_TEST_POSTGRES_ADMIN_URL: postgresql+psycopg://scanner_test:scanner_ci_only@127.0.0.1:5432/postgres SCANNER_REQUIRE_POSTGRES_TESTS: "1" @@ -68,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 diff --git a/tests/test_supply_chain_policy.py b/tests/test_supply_chain_policy.py index 03616d1..d2c584c 100644 --- a/tests/test_supply_chain_policy.py +++ b/tests/test_supply_chain_policy.py @@ -746,10 +746,15 @@ def test_ci_requires_real_postgres_and_application_image_smoke(): The ordinary suite also needs a private file-backed SQLite default: Alembic and application sessions otherwise open different in-memory databases, so a successful upgrade cannot initialize the later session. + GitHub resolves job-level environment before a runner exists. The + runner.temp expression must live on the test step, where that context + is available, or GitHub rejects the workflow before creating any jobs. """ workflow = yaml.safe_load((ROOT / ".github/workflows/quality-and-security.yml").read_text(encoding="utf-8")) verify = workflow["jobs"]["verify"] - assert verify["env"]["DATABASE_URL"] == "sqlite:///${{ runner.temp }}/scanner-default.db" + assert "DATABASE_URL" not in verify["env"] + test_step = next(step for step in verify["steps"] if step["name"] == "Run tests with coverage gate") + assert test_step["env"]["DATABASE_URL"] == "sqlite:///${{ runner.temp }}/scanner-default.db" compose = yaml.safe_load((ROOT / "docker-compose.yml").read_text(encoding="utf-8")) assert verify["services"]["postgres"]["image"] == compose["services"]["postgres"]["image"] assert verify["env"]["SCANNER_REQUIRE_POSTGRES_TESTS"] == "1"