"""Integration: migration 0016 (ui_settings semantic colors) schema contract (phase 93, task 01). Drives the **real Alembic engine** against the live dev database (``podman compose up -d db``), mirroring the house pattern of ``test_migration_0014.py`` (information_schema assertions on the state the migration must leave). The tests target the 0015 → 0016 step explicitly so later migrations cannot break them: * upgrade 0015 → 0016 → the 8 semantic columns exist with the full contract (VARCHAR(7) NULL, no server defaults — the row is created only by the PUT upsert, house rule) while the 0014/0015 columns (``app_name``, ``grid_line``, ``brand_ink``) survive; * an inserted id-1 row round-trips its semantic values (the PUT upsert's shape); * downgrade to 0015 → the 8 columns are GONE (A13 — reversible), the rest of the ``ui_settings`` schema (and ``api_tokens``) survives; * upgrade back to 0016 → the 8 columns are back (round-trip). The ``alembic`` fixture guarantees the DB ends at head even if a test fails or the process is interrupted. """ from __future__ import annotations from collections.abc import Iterator from typing import Any import pytest from alembic.config import Config from sqlalchemy import text from sqlalchemy.orm import Session from alembic import command from app.db import db_available #: The 8 semantic columns (phase 93 — B3 revised): ok / err / accent #: families, VARCHAR(7) NULL ``#rrggbb``, NULL = the built-in (B1). SEMANTIC_COLUMNS = ( "ok_bg", "ok_ink", "err_bg", "err_ink", "err_line", "accent_bg", "accent_ink", "accent_line", ) @pytest.fixture() def alembic(db: Session) -> Iterator[Config]: """Real Alembic config bound to the dev DB (URL from app settings). Starts at head (repairs an interrupted earlier run); teardown upgrades to head no matter what happened, so the dev DB is never left below head. """ if not db_available(): pytest.skip("Postgres not reachable — run `podman compose up -d db` first") cfg = Config() # no alembic.ini file — env.py gets the URL from app config cfg.set_main_option("script_location", "alembic") command.upgrade(cfg, "head") try: yield cfg finally: command.upgrade(cfg, "head") def _version(db: Session) -> str | None: return db.execute(text("SELECT version_num FROM alembic_version")).scalar() def _column(db: Session, table: str, column: str) -> tuple[Any, ...] | None: """(data_type, is_nullable, column_default, character_maximum_length) for one table column.""" row = db.execute( text( "SELECT data_type, is_nullable, column_default, character_maximum_length" " FROM information_schema.columns" " WHERE table_name = :t AND column_name = :c" ), {"t": table, "c": column}, ).fetchone() return tuple(row) if row is not None else None def _insert_row(db: Session) -> None: """Insert the single row (the PUT upsert's shape) with two semantic values set and the rest NULL — the NULL = built-in state the resolver merges.""" db.execute( text( "INSERT INTO ui_settings (id, ok_ink, accent_bg) VALUES (1, :o, :a)" ), {"o": "#444444", "a": "#222222"}, ) db.commit() def _delete_row(db: Session) -> None: db.execute(text("DELETE FROM ui_settings WHERE id = 1")) db.commit() def test_upgrade_to_0016_adds_the_8_semantic_columns( db: Session, alembic: Config ) -> None: """Upgrade 0015 → 0016: the 8 semantic columns exist with the full contract (VARCHAR(7) NULL — NULL = the built-in, B1; no server defaults anywhere: a missing row means "defaults"); all 8 are ABSENT at 0015 and the pre-0016 columns survive the upgrade.""" command.downgrade(alembic, "0015") # start from the pre-0016 state assert _version(db) == "0015" for name in SEMANTIC_COLUMNS: assert _column(db, "ui_settings", name) is None, ( f"ui_settings.{name} must be absent at 0015" ) command.upgrade(alembic, "0016") assert _version(db) == "0016", "alembic_version must be at 0016" for name in SEMANTIC_COLUMNS: col = _column(db, "ui_settings", name) assert col is not None, f"ui_settings.{name} is missing" assert col[0] == "character varying", f"ui_settings.{name} must be VARCHAR" assert col[1] == "YES", f"ui_settings.{name} must be NULL (the built-in, B1)" assert col[2] is None, f"ui_settings.{name} must have no server default" assert col[3] == 7, f"ui_settings.{name} must be String(7) — #rrggbb" # The 0014/0015 columns survive the additive upgrade. for name in ("app_name", "grid_line", "brand_ink"): col = _column(db, "ui_settings", name) assert col is not None, f"ui_settings.{name} must survive the upgrade" def test_inserted_id_1_row_round_trips_semantic_values( db: Session, alembic: Config ) -> None: """At 0016, the single row (id 1, the PUT upsert's shape) round-trips its semantic values verbatim and keeps the unset columns NULL (identity and the other semantic columns).""" command.upgrade(alembic, "head") _insert_row(db) try: row = db.execute( text( "SELECT id, ok_ink, accent_bg, ok_bg, err_ink, accent_line" " FROM ui_settings WHERE id = 1" ) ).fetchone() assert row is not None, "the ui_settings row must exist" assert row[0] == 1, "the single row is always id 1" assert row[1] == "#444444", "ok_ink must round-trip verbatim" assert row[2] == "#222222", "accent_bg must round-trip verbatim" assert row[3] is None, "ok_bg must stay NULL (the built-in, B1)" assert row[4] is None, "err_ink must stay NULL (the built-in, B1)" assert row[5] is None, "accent_line must stay NULL (the built-in, B1)" finally: _delete_row(db) def test_downgrade_to_0015_drops_the_8_columns(db: Session, alembic: Config) -> None: """Downgrade 0016 → 0015: the 8 semantic columns are gone (A13 — fully reversible) while the rest of the schema survives (the 0015 ``grid_line`` column, the 0014 strings, ``api_tokens``).""" command.downgrade(alembic, "0015") assert _version(db) == "0015" for name in SEMANTIC_COLUMNS: assert _column(db, "ui_settings", name) is None, ( f"ui_settings.{name} must be dropped" ) grid = _column(db, "ui_settings", "grid_line") assert grid is not None and grid[3] == 7, ( "grid_line (0015) must survive the downgrade" ) app_name = _column(db, "ui_settings", "app_name") assert app_name is not None and app_name[3] == 300, ( "app_name (0014) must survive the downgrade" ) token_col = _column(db, "api_tokens", "token_hash") assert token_col is not None and token_col[0] == "character varying", ( "api_tokens.token_hash must survive the downgrade" ) def test_upgrade_round_trip_restores_the_columns(db: Session, alembic: Config) -> None: """Downgrade to 0015, then upgrade back to 0016: the 8 columns are back with the column contract intact.""" command.downgrade(alembic, "0015") command.upgrade(alembic, "0016") assert _version(db) == "0016", "round-trip upgrade must land at 0016" for name in SEMANTIC_COLUMNS: col = _column(db, "ui_settings", name) assert col is not None, f"ui_settings.{name} must be back" assert col[0] == "character varying", f"ui_settings.{name} must be VARCHAR" assert col[1] == "YES", f"ui_settings.{name} must be NULL after the round-trip" assert col[3] == 7, f"ui_settings.{name} must be String(7) after the round-trip"