153 lines
6.0 KiB
Python
153 lines
6.0 KiB
Python
"""Integration: migration 0005 (kb_overview) schema contract.
|
|
|
|
Drives the **real Alembic engine** against the live dev database
|
|
(``podman compose up -d db``), mirroring the style of
|
|
``test_migration_0004.py`` (information_schema assertions on the state the
|
|
migration must leave):
|
|
|
|
* upgrade to head → ``kb_overview`` exists with exactly the three columns
|
|
the phase locks in (``id INTEGER PK`` default 1, ``content TEXT NOT NULL``
|
|
default ``''``, ``updated_at TIMESTAMPTZ NOT NULL`` default ``now()``),
|
|
and a bare insert lands the single-row defaults (id=1, content='');
|
|
* downgrade to 0004 → the table is gone;
|
|
* upgrade to head again → it is 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
|
|
|
|
|
|
@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 _column(db: Session, column: str) -> tuple[Any, ...] | None:
|
|
"""(data_type, is_nullable, column_default) for one kb_overview column."""
|
|
row = db.execute(
|
|
text(
|
|
"SELECT data_type, is_nullable, column_default"
|
|
" FROM information_schema.columns"
|
|
" WHERE table_name = 'kb_overview' AND column_name = :c"
|
|
),
|
|
{"c": column},
|
|
).fetchone()
|
|
return tuple(row) if row is not None else None
|
|
|
|
|
|
def _version(db: Session) -> str | None:
|
|
return db.execute(text("SELECT version_num FROM alembic_version")).scalar()
|
|
|
|
|
|
def test_upgrade_to_head_creates_kb_overview(db: Session, alembic: Config) -> None:
|
|
"""Upgrade to head: the single-row table exists with the locked
|
|
column types, nullability, and server defaults."""
|
|
command.downgrade(alembic, "0004") # start from the pre-0005 state
|
|
assert _version(db) == "0004"
|
|
|
|
command.upgrade(alembic, "head")
|
|
assert _version(db) == "0005", "alembic_version must be at 0005 (head)"
|
|
|
|
pk = db.execute(
|
|
text(
|
|
"SELECT column_name FROM information_schema.table_constraints tc"
|
|
" JOIN information_schema.key_column_usage kcu"
|
|
" ON tc.constraint_name = kcu.constraint_name"
|
|
" WHERE tc.table_name = 'kb_overview' AND tc.constraint_type = 'PRIMARY KEY'"
|
|
)
|
|
).scalar()
|
|
assert pk == "id", "kb_overview primary key must be id"
|
|
|
|
id_col = _column(db, "id")
|
|
assert id_col is not None, "kb_overview.id is missing"
|
|
assert id_col[0] == "integer", "kb_overview.id must be INTEGER"
|
|
assert id_col[1] == "NO", "kb_overview.id must be NOT NULL"
|
|
assert id_col[2] == "1", "kb_overview.id must have server default 1"
|
|
|
|
content = _column(db, "content")
|
|
assert content is not None, "kb_overview.content is missing"
|
|
assert content[0] == "text", "kb_overview.content must be TEXT"
|
|
assert content[1] == "NO", "kb_overview.content must be NOT NULL"
|
|
assert content[2] is not None and "''" in content[2], (
|
|
"kb_overview.content must have server default ''"
|
|
)
|
|
|
|
updated = _column(db, "updated_at")
|
|
assert updated is not None, "kb_overview.updated_at is missing"
|
|
assert updated[0] == "timestamp with time zone", "kb_overview.updated_at must be TIMESTAMPTZ"
|
|
assert updated[1] == "NO", "kb_overview.updated_at must be NOT NULL"
|
|
assert updated[2] is not None and "now()" in updated[2], (
|
|
"kb_overview.updated_at must have server default now()"
|
|
)
|
|
|
|
|
|
def test_bare_insert_gets_single_row_defaults(db: Session, alembic: Config) -> None:
|
|
"""A column-less insert lands the single-row shape the phase relies on:
|
|
``id = 1``, ``content = ''``, server-stamped ``updated_at``."""
|
|
command.upgrade(alembic, "head")
|
|
try:
|
|
db.execute(text("DELETE FROM kb_overview"))
|
|
db.execute(text("INSERT INTO kb_overview DEFAULT VALUES"))
|
|
db.commit()
|
|
row = db.execute(
|
|
text("SELECT id, content, updated_at IS NOT NULL FROM kb_overview")
|
|
).fetchone()
|
|
assert row is not None, "the bare insert must land one row"
|
|
assert row[0] == 1, "kb_overview.id must default to 1"
|
|
assert row[1] == "", "kb_overview.content must default to the empty string"
|
|
assert row[2] is True, "kb_overview.updated_at must be stamped by the server"
|
|
finally:
|
|
db.execute(text("DELETE FROM kb_overview"))
|
|
db.commit()
|
|
|
|
|
|
def test_downgrade_to_0004_drops_table(db: Session, alembic: Config) -> None:
|
|
"""Downgrade to 0004: the table is dropped (A13 — reversible)."""
|
|
command.downgrade(alembic, "0004")
|
|
assert _version(db) == "0004"
|
|
|
|
exists = db.execute(
|
|
text("SELECT to_regclass('public.kb_overview') IS NOT NULL")
|
|
).scalar()
|
|
assert exists is False, "kb_overview must be dropped by the downgrade"
|
|
|
|
|
|
def test_upgrade_round_trip_restores_table(db: Session, alembic: Config) -> None:
|
|
"""Upgrade back to head after the downgrade: table + defaults are back."""
|
|
command.upgrade(alembic, "head")
|
|
assert _version(db) == "0005", "round-trip upgrade must land at 0005 (head)"
|
|
|
|
id_col = _column(db, "id")
|
|
assert id_col is not None and id_col[2] == "1", "kb_overview.id must be back with default 1"
|
|
|
|
content = _column(db, "content")
|
|
assert content is not None and content[2] is not None and "''" in content[2], (
|
|
"kb_overview.content must keep its '' default after the round-trip"
|
|
)
|