From 9d548da7d6e3c3f810333825988e0dc15190925b Mon Sep 17 00:00:00 2001 From: Jacob Hagstedt Date: Thu, 17 Sep 2026 19:45:50 +0200 Subject: [PATCH] Guard against schema drift and fields dropped on the wire Two mistakes have recurred in this codebase: the hand-maintained schema in db/connection.py drifting from the Alembic chain, and a field quietly going missing from the hand-written mapping onto the plan response. Tests existed for individual instances of the second, written after each one was found. Both are now checked structurally. One test builds a database from the migrations and another from the hand-written schema and compares their real shape, ignoring cosmetic column ordering but failing on a missing column, wrong nullability, absent foreign key or index. The other walks the proposal dataclass and asserts every field reaches the response, so a field added later is covered without anyone remembering to extend the test. Anything deliberately kept off the wire goes in an explicit allow-list. Both were confirmed by breaking what they guard and watching them name it. --- .../tests/integration/test_schema_parity.py | 251 +++++++++++++++++ backend/tests/unit/test_plan_router.py | 259 ++++++++++++++++++ 2 files changed, 510 insertions(+) create mode 100644 backend/tests/integration/test_schema_parity.py diff --git a/backend/tests/integration/test_schema_parity.py b/backend/tests/integration/test_schema_parity.py new file mode 100644 index 0000000..aee2b5a --- /dev/null +++ b/backend/tests/integration/test_schema_parity.py @@ -0,0 +1,251 @@ +"""Mechanism test: Alembic migrations vs. `_FULL_SCHEMA` must describe one schema. + +Production boots via `run_migrations()` (the real Alembic chain in +`backend/migrations/versions/`). All 518+ existing tests instead get their schema +from `ensure_schema()`, which applies `_FULL_SCHEMA` — a hand-maintained DDL string +in `app/db/connection.py` — directly, bypassing Alembic entirely (see +`tests/conftest.py::connection`). + +Nothing has ever compared the two. They have drifted before (see +`PLANNING_ENGINE_SPEC.md` §10.5: "Do not forget `_FULL_SCHEMA`" is called out +precisely because someone forgot it once already). When they drift, every test +in the suite keeps passing against a schema production does not have, and the +gap is only discovered when production breaks. + +This test builds one SQLite database via each path and asserts the resulting +schemas are structurally equivalent: same tables, same columns (name, type, +nullability, default, primary-key membership), same CHECK constraints, same +foreign keys, same indexes/uniqueness. Column *order* and cosmetic DDL +formatting are deliberately ignored — `_FULL_SCHEMA` writes each table's final +column layout by hand, while the migration chain appends columns via `ALTER +TABLE ... ADD COLUMN`, so the two legitimately end up in different column +orders for the same set of columns. That is not a bug and must not fail this +test. +""" + +from __future__ import annotations + +import asyncio +import re +import sqlite3 +from pathlib import Path + +import aiosqlite +import pytest + +from app.db.connection import ensure_schema, run_migrations + +# Bookkeeping tables that exist only because of *how* a schema was applied, +# never part of the application's own data model. +_IGNORED_TABLES = {"sqlite_sequence", "alembic_version"} + + +def _is_app_table(name: str) -> bool: + return name not in _IGNORED_TABLES and not name.startswith("sqlite_") + + +def _table_names(cur: sqlite3.Cursor) -> set[str]: + rows = cur.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall() + return {row[0] for row in rows if _is_app_table(row[0])} + + +def _normalize_type(sql_type: str | None) -> str: + return (sql_type or "").strip().upper() + + +def _normalize_default(default: object) -> str | None: + if default is None: + return None + return str(default).strip() + + +def _columns(cur: sqlite3.Cursor, table: str) -> dict[str, tuple[str, bool, str | None, int]]: + """name -> (normalized type, notnull, normalized default, pk ordinal).""" + rows = cur.execute(f'PRAGMA table_info("{table}")').fetchall() + # row: cid, name, type, notnull, dflt_value, pk + return { + row[1]: (_normalize_type(row[2]), bool(row[3]), _normalize_default(row[4]), row[5]) + for row in rows + } + + +def _foreign_keys(cur: sqlite3.Cursor, table: str) -> set[tuple[str, str, str, str, str]]: + rows = cur.execute(f'PRAGMA foreign_key_list("{table}")').fetchall() + # row: id, seq, table, from, to, on_update, on_delete, match + return {(row[2], row[3], row[4] or "", row[5], row[6]) for row in rows} + + +def _indexes(cur: sqlite3.Cursor, table: str) -> set[tuple[bool, tuple[str, ...]]]: + """(is_unique, ordered column names) per index — index *names* are not compared, + they are SQLite-generated (`sqlite_autoindex_...`) and carry no meaning of their own. + """ + index_rows = cur.execute(f'PRAGMA index_list("{table}")').fetchall() + # row: seq, name, unique, origin, partial + result: set[tuple[bool, tuple[str, ...]]] = set() + for row in index_rows: + index_name, is_unique = row[1], bool(row[2]) + info_rows = cur.execute(f'PRAGMA index_info("{index_name}")').fetchall() + # row: seqno, cid, name — order by seqno for a stable column tuple. + cols = tuple(c[2] for c in sorted(info_rows, key=lambda c: c[0])) + result.add((is_unique, cols)) + return result + + +def _extract_checks(create_table_sql: str) -> list[str]: + """Pull out each balanced `CHECK ( ... )` clause's inner text, raw.""" + checks: list[str] = [] + upper = create_table_sql.upper() + pos = 0 + while True: + found = upper.find("CHECK", pos) + if found == -1: + break + paren_start = create_table_sql.find("(", found) + if paren_start == -1: + break + depth = 0 + end = paren_start + for end in range(paren_start, len(create_table_sql)): + char = create_table_sql[end] + if char == "(": + depth += 1 + elif char == ")": + depth -= 1 + if depth == 0: + break + checks.append(create_table_sql[paren_start + 1 : end]) + pos = end + 1 + return checks + + +def _normalize_check(check_text: str) -> str: + return re.sub(r"\s+", " ", check_text).strip() + + +def _checks(cur: sqlite3.Cursor, table: str) -> frozenset[str]: + row = cur.execute( + "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?", (table,) + ).fetchone() + sql = row[0] if row and row[0] else "" + return frozenset(_normalize_check(c) for c in _extract_checks(sql)) + + +def _diff_schemas(migrated: sqlite3.Connection, full_schema: sqlite3.Connection) -> list[str]: + """Return a list of human-readable mismatch descriptions; empty means equivalent.""" + diffs: list[str] = [] + migrated_cur = migrated.cursor() + full_schema_cur = full_schema.cursor() + + migrated_tables = _table_names(migrated_cur) + full_schema_tables = _table_names(full_schema_cur) + + only_in_migrations = migrated_tables - full_schema_tables + only_in_full_schema = full_schema_tables - migrated_tables + if only_in_migrations: + diffs.append( + f"tables only produced by run_migrations() (missing from _FULL_SCHEMA): " + f"{sorted(only_in_migrations)}" + ) + if only_in_full_schema: + diffs.append( + f"tables only produced by _FULL_SCHEMA (missing from the migration chain): " + f"{sorted(only_in_full_schema)}" + ) + + for table in sorted(migrated_tables & full_schema_tables): + migrated_cols = _columns(migrated_cur, table) + full_schema_cols = _columns(full_schema_cur, table) + + missing_in_full_schema = migrated_cols.keys() - full_schema_cols.keys() + missing_in_migrations = full_schema_cols.keys() - migrated_cols.keys() + if missing_in_full_schema: + diffs.append( + f"table '{table}': column(s) {sorted(missing_in_full_schema)} exist after " + f"run_migrations() but are missing from _FULL_SCHEMA" + ) + if missing_in_migrations: + diffs.append( + f"table '{table}': column(s) {sorted(missing_in_migrations)} exist in " + f"_FULL_SCHEMA but the migration chain never creates them" + ) + + for col in sorted(migrated_cols.keys() & full_schema_cols.keys()): + m_type, m_notnull, m_default, m_pk = migrated_cols[col] + f_type, f_notnull, f_default, f_pk = full_schema_cols[col] + if (m_type, m_notnull, m_default, m_pk) != (f_type, f_notnull, f_default, f_pk): + diffs.append( + f"table '{table}' column '{col}': migrations gives " + f"(type={m_type!r}, notnull={m_notnull}, default={m_default!r}, pk={m_pk}) " + f"but _FULL_SCHEMA gives " + f"(type={f_type!r}, notnull={f_notnull}, default={f_default!r}, pk={f_pk})" + ) + + m_checks = _checks(migrated_cur, table) + f_checks = _checks(full_schema_cur, table) + if m_checks != f_checks: + diffs.append( + f"table '{table}': CHECK constraints differ — " + f"migrations only: {sorted(m_checks - f_checks)}; " + f"_FULL_SCHEMA only: {sorted(f_checks - m_checks)}" + ) + + m_fks = _foreign_keys(migrated_cur, table) + f_fks = _foreign_keys(full_schema_cur, table) + if m_fks != f_fks: + diffs.append( + f"table '{table}': foreign keys differ (ref_table, from_col, to_col, " + f"on_update, on_delete) — " + f"migrations only: {sorted(m_fks - f_fks)}; " + f"_FULL_SCHEMA only: {sorted(f_fks - m_fks)}" + ) + + m_idx = _indexes(migrated_cur, table) + f_idx = _indexes(full_schema_cur, table) + if m_idx != f_idx: + diffs.append( + f"table '{table}': indexes differ (is_unique, columns) — " + f"migrations only: {sorted(m_idx - f_idx)}; " + f"_FULL_SCHEMA only: {sorted(f_idx - m_idx)}" + ) + + return diffs + + +async def _apply_full_schema(db_path: Path) -> None: + conn = await aiosqlite.connect(db_path) + try: + await ensure_schema(conn) + finally: + await conn.close() + + +def test_alembic_migrations_and_full_schema_agree(tmp_path: Path) -> None: + """`run_migrations()` (production) and `ensure_schema()` (tests) must build the + same schema. A failure here means the 518+ tests using `_FULL_SCHEMA` are + exercising a schema production does not actually have — see module docstring. + """ + migrated_db = tmp_path / "via_migrations.db" + full_schema_db = tmp_path / "via_full_schema.db" + + # Alembic chain, exactly as production's app lifespan runs it. + run_migrations(f"sqlite:///{migrated_db}") + + # `ensure_schema()` is async (an aiosqlite connection); `run_migrations()` above + # must run to completion first since it drives its own `asyncio.run()` internally + # and cannot be called from within a running event loop. + asyncio.run(_apply_full_schema(full_schema_db)) + + migrated_conn = sqlite3.connect(migrated_db) + full_schema_conn = sqlite3.connect(full_schema_db) + try: + diffs = _diff_schemas(migrated_conn, full_schema_conn) + finally: + migrated_conn.close() + full_schema_conn.close() + + if diffs: + pytest.fail( + "Schema drift between Alembic migrations (production) and _FULL_SCHEMA " + "(tests) in backend/app/db/connection.py:\n- " + "\n- ".join(diffs), + pytrace=False, + ) diff --git a/backend/tests/unit/test_plan_router.py b/backend/tests/unit/test_plan_router.py index 1403921..d90b580 100644 --- a/backend/tests/unit/test_plan_router.py +++ b/backend/tests/unit/test_plan_router.py @@ -15,6 +15,7 @@ Uses the same in-memory sqlite `connection` fixture as the rest of the suite graph without touching the real `foraldraledig.db` file or running Alembic. """ +import dataclasses from datetime import date, datetime from uuid import uuid4 @@ -35,6 +36,7 @@ from app.repositories_impl.sqlite.parental_leave_repository import SQLiteParenta from app.repositories_impl.sqlite.personal_data_repository import SQLitePersonalDataRepository from app.repositories_impl.sqlite.vacation_days_repository import SQLiteVacationDaysRepository from app.repositories_impl.sqlite.workspace_repository import SQLiteWorkspaceRepository +from app.services.allocation_service import PlanProposal, WindowOutcome def _make_fetch( @@ -809,3 +811,260 @@ class TestLeaveWishWindowModeOverTheWire: assert outcome["strategy"] == "unplaced" assert outcome["weeks_funded"] == 0.0 assert outcome["shortfall_days"] == 7 + + +# --------------------------------------------------------------------------- +# Systematic wire-contract guard for the plan proposal (mechanism test) +# --------------------------------------------------------------------------- +# +# The classes above each guard exactly one field that was already silently +# dropped from `app/models/plan.py`'s hand-written `from_domain()` mapping +# once (surplus_advice, age4_projections, wish_name_message_key). That is +# whack-a-mole: it only ever protects a field *after* losing it has already +# been painful enough to write a regression test for. +# +# This class instead walks `dataclasses.fields()` of the domain `PlanProposal` +# (and its nested `WindowOutcome`) and checks each one has a surviving +# counterpart in the actual HTTP JSON. Critically, a domain field with no +# entry in the rename map or allow-list below is checked under its OWN name +# by default -- so a brand-new field added to `PlanProposal` (or +# `WindowOutcome`) and forgotten in `from_domain()` fails HERE, without this +# test file needing to be touched first. Only intentional renames/fan-outs +# and intentional exclusions require a (documented) entry. + +# Domain field name -> wire field name(s) on `PlanProposalResponse`. Only +# fields whose wire shape differs from a straight "same name" passthrough +# need an entry. `balances_after` is the one case today: a single nested +# `PlanBalances` dataclass that `from_domain()` fans out into two separate +# top-level wire lists. +_PLAN_PROPOSAL_RENAMES: dict[str, tuple[str, ...]] = { + "balances_after": ("pool_balances", "vacation_balances"), +} + +# Domain fields on `PlanProposal` deliberately absent from the wire, with why. +# Empty today: every `PlanProposal` field currently reaches the HTTP response. +_PLAN_PROPOSAL_ALLOWLIST: dict[str, str] = {} + +# Same idea for `WindowOutcome` -> `WindowOutcomeResponse`: every field there +# is a straight same-name passthrough today, so both maps are empty, but they +# stay explicit (not just "no map at all") so the intent -- and the place to +# add the next rename/exclusion -- is obvious. +_WINDOW_OUTCOME_RENAMES: dict[str, tuple[str, ...]] = {} +_WINDOW_OUTCOME_ALLOWLIST: dict[str, str] = {} + + +def _wire_keys_for(domain_field: str, renames: dict[str, tuple[str, ...]]) -> tuple[str, ...]: + """The wire key(s) `domain_field` must appear as, absent an explicit rename.""" + return renames.get(domain_field, (domain_field,)) + + +def _assert_domain_fields_reach_the_wire( + dataclass_type: type, + renames: dict[str, tuple[str, ...]], + allowlist: dict[str, str], + wire_obj: dict[str, object], + *, + context: str, + check_nonempty: bool = True, +) -> None: + """Fail loudly, naming the exact field, if any `dataclass_type` field has + no live counterpart in `wire_obj` (a decoded JSON object from the HTTP + response) -- unless it is explicitly excused via `allowlist`. + + `check_nonempty` additionally requires list-typed wire fields to be + non-empty. Leave it on for a single, richly-seeded top-level object (e.g. + the `PlanProposalResponse` body, where every list should have content); + turn it off when walking a list of *instances* of a nested dataclass + (e.g. each individual `WindowOutcomeResponse`) — a specific outcome's own + list field (like `remedies`) can be legitimately empty (a fully-funded + window has no remedies) without that being a dropped-field bug. + """ + for f in dataclasses.fields(dataclass_type): + if f.name in allowlist: + continue + for wire_key in _wire_keys_for(f.name, renames): + assert wire_key in wire_obj, ( + f"{context}: domain field '{f.name}' has no counterpart " + f"'{wire_key}' in the wire response. Either from_domain() in " + f"app/models/plan.py silently dropped it, or (if '{f.name}' is " + f"a brand-new PlanProposal/WindowOutcome field) it needs a " + f"wire mapping AND a rename/allow-list entry in this test." + ) + if not check_nonempty: + continue + value = wire_obj[wire_key] + if isinstance(value, list): + assert value, ( + f"{context}: wire field '{wire_key}' (from domain field " + f"'{f.name}') is present but empty. Either the seeded " + f"workspace in this test needs to be richer, or " + f"from_domain() is silently producing an empty list where " + f"the domain data is not actually empty." + ) + + +class TestPlanProposalFieldParityWithDomain: + """Mechanism test: every `PlanProposal`/`WindowOutcome` field must survive + the hand-written `PlanProposalResponse.from_domain()` mapping, not just + the three fields the classes above already know to check for by name. + """ + + async def test_every_plan_proposal_and_window_outcome_field_reaches_the_wire( + self, + client: AsyncClient, + connection: aiosqlite.Connection, + ) -> None: + now = datetime.now() + workspace_repo = SQLiteWorkspaceRepository(connection) + personal_data_repo = SQLitePersonalDataRepository(connection) + child_repo = SQLiteChildRepository(connection) + parental_leave_repo = SQLiteParentalLeaveRepository(connection) + vacation_repo = SQLiteVacationDaysRepository(connection) + + ws = await workspace_repo.create(Workspace(id=uuid4(), name="Familj", created_at=now)) + parent = await personal_data_repo.create( + PersonalData( + uuid=uuid4(), + workspace_id=ws.id, + name="Alex", + salary_type="monthly", + created_at=now, + ) + ) + + # Child A: pool fully exhausted + no vacation in the wish's funding + # year -> a guaranteed shortfall (populates outcomes[].remedies, + # outcome.data, outcome.cost_breakdown with a genuinely non-trivial + # story) — same recipe as TestProposeMessageKeysOverTheWire above. + child_shortfall = await child_repo.create( + Child( + id=uuid4(), + workspace_id=ws.id, + name="Barn Ett", + birth_date=date.today() - relativedelta(months=6), + created_at=now, + ) + ) + await parental_leave_repo.create( + _make_fetch(parent.uuid, ws.id, child_shortfall.id, sjukpenning_left=0, lagsta_left=0) + ) + + next_year = date.today().year + 1 + await vacation_repo.create( + VacationDays( + id=uuid4(), + personal_data_uuid=parent.uuid, + workspace_id=ws.id, + year=next_year, + total_days=0, + saved_days_in=0, + ) + ) + # Spare vacation *this* year gives `carry_vacation_forward` something + # to point at: available (25) above the semesterlagen minimum (20). + await vacation_repo.create( + VacationDays( + id=uuid4(), + personal_data_uuid=parent.uuid, + workspace_id=ws.id, + year=date.today().year, + total_days=25, + saved_days_in=0, + ) + ) + + preset_wish = await client.post( + "/api/leave-wishes", + json={ + "workspace_id": str(ws.id), + "child_id": str(child_shortfall.id), + "name": "Vinter", + "name_message_key": "wishEditor.presetVinter", + "window_start_month": 12, + "window_start_day": 1, + "window_end_month": 12, + "window_end_day": 7, + "target_weeks": 1, + "recurrence": "once", + "first_year": next_year, + }, + ) + assert preset_wish.status_code == 201 + + # Child B: no `parental_leave_fetch` row at all -> the full default FP + # pool, untouched by the small wish below, still swamps the age-4 + # ceiling -> guaranteed `surplus_advice` + `days_lost_at_age_4` + # warning + a real `age4_projections` story — same recipe as + # TestProposeSurplusAdviceOverTheWire above. The small wish itself + # (well inside the huge default pool) also guarantees `days` and a + # second, fully-funded `outcomes` entry are non-trivially populated. + child_surplus = await child_repo.create( + Child( + id=uuid4(), + workspace_id=ws.id, + name="Barn Två", + birth_date=date.today() - relativedelta(months=6), + created_at=now, + ) + ) + funded_wish = await client.post( + "/api/leave-wishes", + json={ + "workspace_id": str(ws.id), + "child_id": str(child_surplus.id), + "name": "Sportlov", + "window_start_month": 2, + "window_start_day": 1, + "window_end_month": 2, + "window_end_day": 7, + "target_weeks": 1, + "recurrence": "once", + "first_year": next_year, + }, + ) + assert funded_wish.status_code == 201 + + response = await client.post("/api/plan/propose", json={"workspace_id": str(ws.id)}) + assert response.status_code == 200 + body = response.json() + + # Sanity-check the seed actually produced a non-trivial proposal + # before trusting the generic field walk below to mean anything. + assert body["days"], "seed should have produced at least one funded day" + assert body["outcomes"], "seed should have produced at least one outcome" + assert body["warnings"], "seed should have produced at least one warning" + assert body["surplus_advice"], "seed should have produced surplus advice" + assert body["age4_projections"], "seed should have produced age4 projections" + assert body["pool_balances"], "seed should have produced pool balances" + assert body["vacation_balances"], "seed should have produced vacation balances" + + _assert_domain_fields_reach_the_wire( + PlanProposal, + _PLAN_PROPOSAL_RENAMES, + _PLAN_PROPOSAL_ALLOWLIST, + body, + context="PlanProposal -> PlanProposalResponse", + ) + + for outcome in body["outcomes"]: + _assert_domain_fields_reach_the_wire( + WindowOutcome, + _WINDOW_OUTCOME_RENAMES, + _WINDOW_OUTCOME_ALLOWLIST, + outcome, + context="WindowOutcome -> WindowOutcomeResponse", + # A specific outcome's own list field (e.g. `remedies`) can be + # legitimately empty (a fully-funded window has none) -- + # presence is still checked, only the blanket emptiness + # assertion is skipped here. See docstring above. + check_nonempty=False, + ) + + # `remedies` is a list-of-dataclass field inside WindowOutcome too — + # confirm the shortfall outcome specifically has a non-empty one + # (the generic walk above only guarantees each outcome's own + # `remedies` key exists and is non-empty where non-empty is expected; + # this pins down that the shortfall scenario, not the funded one, is + # the one carrying it). + shortfall_outcome = next(o for o in body["outcomes"] if o["shortfall_days"] > 0) + assert shortfall_outcome["remedies"], "expected the carry_vacation_forward remedy"