PacketStation/tests/server/test_schema.py

"""Migration 0003_domain: enums, constraints, append-only tables, views and app-role grants."""

from __future__ import annotations

import uuid
from typing import Any

import psycopg
import pytest
from pshelpers import app_conn, needs_pg
from psycopg import sql

from packetstation import db, migrate
from packetstation.settings import Settings

pytestmark = needs_pg

ENUMS = {
    "packet_status": [
        "draft",
        "warehouse",
        "ready",
        "assigned",
        "in_progress",
        "review",
        "qc",
        "complete",
        "filed",
    ],
    "wh_stage": ["incoming", "pulling", "hold", "pulled"],
    "co_review": ["pending", "approved", "rejected"],
    "co_origin": ["visual", "drawing", "manual"],
    "source_kind": ["visual", "hand"],
    "project_status": ["open", "hold", "closed"],
    "route_kind": ["databook", "shipping"],
    "drawing_state": [
        "none",
        "needed",
        "attached",
        "rev_mismatch",
        "not_released",
        "not_found",
        "lookup_failed",
        "hand",
    ],
    "itp_status": ["open", "signed", "reopened", "complete"],
}

# One minimal INSERT per append-only table, with ids from seed().
APPEND_INSERTS = {
    "project_renumbers": (
        "INSERT INTO project_renumbers (project_id, old_number, new_number, by) "
        "VALUES (%(project)s, 'L998', 'L999', %(user)s)"
    ),
    "short_receipts": (
        "INSERT INTO short_receipts (short_id, packet_id, qty, by) VALUES (%(short)s, %(packet)s, 1, %(user)s)"
    ),
    "packet_transitions": (
        "INSERT INTO packet_transitions (packet_id, action, by) VALUES (%(packet)s, 'send_to_warehouse', %(user)s)"
    ),
    "qc_history": "INSERT INTO qc_history (packet_id, result) VALUES (%(packet)s, 'accepted')",
    "packet_log": (
        "INSERT INTO packet_log (packet_id, action, by) VALUES (%(packet)s, 'Packet created by hand', %(user)s)"
    ),
    "itp_rejects": "INSERT INTO itp_rejects (itp_id, line_key, ncr) VALUES (%(itp)s, 'k1', 'NCR-1')",
    "itp_log": "INSERT INTO itp_log (itp_id, action, by) VALUES (%(itp)s, 'ITP started', %(user)s)",
    "co_verifications": (
        "INSERT INTO co_verifications (co_id, dept_code, by, signature_id) VALUES (%(co)s, 'WH', %(user)s, %(sig)s)"
    ),
    "print_log": (
        "INSERT INTO print_log (packet_id, kind, what, by) VALUES (%(packet)s, 'traveler', 'Job traveler', %(user)s)"
    ),
}
APPEND_ONLY = list(APPEND_INSERTS)


def q(template: str, name: str) -> sql.Composed:
    return sql.SQL(template).format(sql.Identifier(name))


def one(conn: Any, query: Any, params: Any = None) -> Any:
    row = conn.execute(query, params).fetchone()
    assert row is not None
    return row[0]


def seed(conn: Any) -> dict[str, Any]:
    """A user, a signature, project L999 with unit 001, one packet with a BOM line, a short, a CO and an ITP."""
    ids: dict[str, Any] = {}
    ids["user"] = uid = uuid.uuid4()
    conn.execute(
        "INSERT INTO users (id, sam_account, display_name, welder_stamp) VALUES (%s, %s, %s, %s)",
        (uid, f"weld.{uid.hex[:6]}", "Andy Welder", "A7"),
    )
    ids["sig"] = one(conn, "INSERT INTO signatures (user_id, action) VALUES (%s, 'test') RETURNING id", (uid,))
    ids["project"] = pid = one(
        conn,
        "INSERT INTO projects (number, name, due, created_by) VALUES ('L999', 'Test', '2026-12-01', %s) RETURNING id",
        (uid,),
    )
    ids["unit"] = unit = one(conn, "INSERT INTO units (project_id, lot) VALUES (%s, '001') RETURNING id", (pid,))
    ids["packet"] = pk = one(
        conn,
        "INSERT INTO packets (unit_id, dept_code, op, created_by) VALUES (%s, '03', 'WELD-PART', %s) RETURNING id",
        (unit, uid),
    )
    ids["bom"] = bom = one(
        conn, "INSERT INTO bom_lines (packet_id, item, qty, part_id) VALUES (%s, 1, 2, '00123') RETURNING id", (pk,)
    )
    ids["short"] = one(
        conn,
        "INSERT INTO bom_shorts (packet_id, bom_line_id, qty, qty_open) VALUES (%s, %s, 2, 2) RETURNING id",
        (pk, bom),
    )
    ids["co"] = new_co(conn, pk, 1, "manual", "approved")
    ids["itp"] = one(conn, "INSERT INTO itps (unit_id, template) VALUES (%s, '{}') RETURNING id", (unit,))
    return ids


def new_co(conn: Any, packet_id: int, co_no: int, origin: str, review: str, hold: bool = False) -> int:
    co_id: int = one(
        conn,
        "INSERT INTO change_orders (packet_id, co_no, origin, review, hold) VALUES (%s, %s, %s, %s, %s) RETURNING id",
        (packet_id, co_no, origin, review, hold),
    )
    return co_id


def test_migrations_apply_from_zero(pg_settings: Settings) -> None:
    with db.connect_as("migrate", pg_settings, autocommit=True) as conn:
        assert migrate.run(conn, db.MIGRATIONS_DIR) == []
        names = [r[0] for r in conn.execute("SELECT name FROM schema_migrations ORDER BY name").fetchall()]
    assert names[:3] == ["0001_base.sql", "0002_identity.sql", "0003_domain.sql"]


def test_enum_values_in_order(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        for name, values in ENUMS.items():
            rows = conn.execute(
                "SELECT e.enumlabel FROM pg_enum e JOIN pg_type t ON t.oid = e.enumtypid "
                "WHERE t.typname = %s ORDER BY e.enumsortorder",
                (name,),
            ).fetchall()
            assert [r[0] for r in rows] == values, name
        # Declaration order is workflow order, so comparisons work.
        assert one(conn, "SELECT 'qc'::packet_status > 'review'::packet_status") is True


def test_departments_weld(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        rows = conn.execute("SELECT code FROM departments WHERE weld ORDER BY code").fetchall()
    assert [r[0] for r in rows] == ["03", "04", "05", "06", "07", "10"]


def test_append_only_tables_refuse_update_and_delete(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        for table in APPEND_ONLY:
            conn.execute(APPEND_INSERTS[table], ids)
            assert one(conn, q("SELECT count(*) FROM {}", table)) == 1, table
            # The app role lacks the privilege.
            for stmt in ("UPDATE {} SET at = at", "DELETE FROM {}", "TRUNCATE {}"):
                with pytest.raises(psycopg.errors.InsufficientPrivilege):
                    conn.execute(q(stmt, table))
    # The owner has the privilege, so the trigger is what refuses it.
    with db.connect_as("migrate", pg_settings, autocommit=True) as conn:
        for table in APPEND_ONLY:
            for stmt in ("UPDATE {} SET at = at", "DELETE FROM {}", "TRUNCATE {} CASCADE"):
                with pytest.raises(psycopg.errors.InsufficientPrivilege, match="append-only"):
                    conn.execute(q(stmt, table))


def test_append_only_set_matches_triggers(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        rows = conn.execute(
            "SELECT DISTINCT c.relname FROM pg_trigger t JOIN pg_class c ON c.oid = t.tgrelid "
            "JOIN pg_proc p ON p.oid = t.tgfoid WHERE p.proname = 'ps_append_only'"
        ).fetchall()
    assert {r[0] for r in rows} == {*APPEND_ONLY, "audit_events", "signin_attempts", "signatures"}


def test_packets_visual_key_live_only(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        ins = (
            "INSERT INTO packets (unit_id, dept_code, source_kind, visual_key, removed_at) "
            "VALUES (%s, '03', 'visual', 'L999001|03|D-100', %s)"
        )
        conn.execute(ins, (ids["unit"], "2026-10-01T00:00:00Z"))
        conn.execute(ins, (ids["unit"], "2026-10-02T00:00:00Z"))
        conn.execute(ins, (ids["unit"], None))
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute(ins, (ids["unit"], None))
        # Packets without a key never collide.
        conn.execute("INSERT INTO packets (unit_id, dept_code) VALUES (%s, '03')", (ids["unit"],))
        conn.execute("INSERT INTO packets (unit_id, dept_code) VALUES (%s, '03')", (ids["unit"],))


def test_soft_delete_unique_indexes(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute("INSERT INTO projects (number) VALUES ('l999')")
        conn.execute("UPDATE projects SET removed_at = now() WHERE id = %s", (ids["project"],))
        conn.execute("INSERT INTO projects (number) VALUES ('l999')")
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute("INSERT INTO units (project_id, lot) VALUES (%s, '001')", (ids["project"],))
        weld = "INSERT INTO welds (packet_id, weld_no, removed_at) VALUES (%s, 1, %s)"
        conn.execute(weld, (ids["packet"], "2026-10-01T00:00:00Z"))
        conn.execute(weld, (ids["packet"], None))
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute(weld, (ids["packet"], None))
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute(
                "INSERT INTO bom_shorts (packet_id, bom_line_id, qty, qty_open) VALUES (%s, %s, 1, 1)",
                (ids["packet"], ids["bom"]),
            )


def test_unit_lot_is_text(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        conn.execute("INSERT INTO units (project_id, lot) VALUES (%s, '7A')", (ids["project"],))
        conn.execute("INSERT INTO units (project_id, lot) VALUES (%s, '010')", (ids["project"],))
        for bad in ("7", "7a", "0001", ""):
            with pytest.raises(psycopg.errors.CheckViolation):
                conn.execute("INSERT INTO units (project_id, lot) VALUES (%s, %s)", (ids["project"], bad))
        numbers = [r[0] for r in conn.execute("SELECT number FROM v_units ORDER BY id").fetchall()]
    assert numbers == ["L999001", "L9997A", "L999010"]


def test_v_units(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        conn.execute("INSERT INTO units (project_id, lot, due) VALUES (%s, '002', '2026-11-15')", (ids["project"],))
        rows = conn.execute(
            "SELECT lot, number, project_number, project_name, project_status::text, due_effective::text "
            "FROM v_units ORDER BY lot"
        ).fetchall()
    assert rows == [
        ("001", "L999001", "L999", "Test", "open", "2026-12-01"),
        ("002", "L999002", "L999", "Test", "open", "2026-11-15"),
    ]


def test_v_open_cos(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        co, pk = ids["co"], ids["packet"]
        verify = "INSERT INTO co_verifications (co_id, dept_code, by, signature_id) VALUES (%s, %s, %s, %s)"
        conn.execute("INSERT INTO co_required_depts (co_id, dept_code) VALUES (%s, 'WH'), (%s, '03')", (co, co))
        conn.execute(verify, (co, "WH", ids["user"], ids["sig"]))
        held = new_co(conn, pk, 2, "visual", "pending", hold=True)
        new_co(conn, pk, 3, "manual", "pending", hold=False)  # 0004: one pending feed CO, visual or drawing
        rejected = new_co(conn, pk, 4, "manual", "rejected", hold=True)
        conn.execute("INSERT INTO co_required_depts (co_id, dept_code) VALUES (%s, '03')", (rejected,))
        rows = conn.execute(
            "SELECT packet_id, co_id, co_no, review::text, hold, pending_depts FROM v_open_cos ORDER BY co_no"
        ).fetchall()
        assert rows == [(pk, co, 1, "approved", False, ["03"]), (pk, held, 2, "pending", True, [])]
        # Once the last department verifies, the approved CO is no longer open.
        conn.execute(verify, (co, "03", ids["user"], ids["sig"]))
        assert [r[0] for r in conn.execute("SELECT co_no FROM v_open_cos").fetchall()] == [2]


def test_v_open_cos_pending_depts_sorted(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        co = ids["co"]
        conn.execute(
            "INSERT INTO co_required_depts (co_id, dept_code) VALUES (%s, 'WH'), (%s, 'QC'), (%s, '06')", (co, co, co)
        )
        assert one(conn, "SELECT pending_depts FROM v_open_cos WHERE co_id = %s", (co,)) == ["06", "QC", "WH"]


def test_one_pending_visual_co_per_packet(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        new_co(conn, ids["packet"], 2, "visual", "pending")
        new_co(conn, ids["packet"], 3, "visual", "approved")
        with pytest.raises(psycopg.errors.UniqueViolation):
            new_co(conn, ids["packet"], 4, "visual", "pending")
        with pytest.raises(psycopg.errors.UniqueViolation):
            new_co(conn, ids["packet"], 4, "drawing", "pending")  # 0004: drawing-origin counts too
        with pytest.raises(psycopg.errors.UniqueViolation):
            new_co(conn, ids["packet"], 2, "manual", "approved")


def test_co_verifications_insert_only(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        conn.execute(APPEND_INSERTS["co_verifications"], ids)
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute(APPEND_INSERTS["co_verifications"], ids)
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute("UPDATE co_verifications SET note = 'x'")
        with pytest.raises(psycopg.errors.NotNullViolation):
            conn.execute(
                "INSERT INTO co_verifications (co_id, dept_code, by) VALUES (%s, '03', %s)", (ids["co"], ids["user"])
            )


def test_packet_touch_sets_updated_at_only(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        conn.execute("UPDATE packets SET title = 'Frame' WHERE id = %s", (ids["packet"],))
        row = conn.execute(
            "SELECT version, updated_at > created_at, updated_at > now() - interval '1 minute' FROM packets "
            "WHERE id = %s",
            (ids["packet"],),
        ).fetchone()
        assert row == (1, True, True)
        # Even an explicit old value is replaced: the trigger owns updated_at.
        conn.execute("UPDATE packets SET updated_at = '2000-01-01T00:00:00Z' WHERE id = %s", (ids["packet"],))
        assert one(conn, "SELECT updated_at > '2001-01-01' FROM packets WHERE id = %s", (ids["packet"],)) is True


def test_app_can_write_every_table_and_read_every_view(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        ids["station"] = one(conn, "SELECT id FROM stations WHERE dept_code = '03' ORDER BY sort LIMIT 1")
        ids["asset"] = one(
            conn,
            "INSERT INTO assets (sha256, size, mime, storage_key, pages, uploaded_by) "
            "VALUES ('\\x00', 10, 'application/pdf', 'ab/abc', 1, %(user)s) RETURNING id",
            ids,
        )
        ids["stage"] = one(
            conn,
            "INSERT INTO packet_stages (packet_id, dept_code, station_id, welder_id) "
            "VALUES (%(packet)s, '03', %(station)s, %(user)s) RETURNING id",
            ids,
        )
        statements = [
            "INSERT INTO user_station_limits (user_id, station_id) VALUES (%(user)s, %(station)s)",
            "INSERT INTO project_aliases (alias, project_id, unit_id, kind) "
            "VALUES ('L998001', %(project)s, %(unit)s, 'import')",
            "INSERT INTO co_items (co_id, old_page, new_page, added) VALUES (%(co)s, 1, 2, true)",
            "INSERT INTO co_required_depts (co_id, dept_code) VALUES (%(co)s, 'QC')",
            "INSERT INTO packet_work_orders (packet_id, wo_base, lot, split, sub, seq) "
            "VALUES (%(packet)s, '0670001', '7A', '0', '0', 10)",
            "INSERT INTO packet_pages (packet_id, n, type, asset_id, co_id) "
            "VALUES (%(packet)s, 1, 'drawing', %(asset)s, %(co)s)",
            "INSERT INTO welds (packet_id, weld_no, items, welder_id, dept_code, stage_id, by) "
            "VALUES (%(packet)s, 1, '{1,2}', %(user)s, '03', %(stage)s, %(user)s)",
            "INSERT INTO heat_tags (packet_id, tag_no, page, item, fx, fy) VALUES (%(packet)s, 1, 1, 1, 0.5, 0.5)",
            "INSERT INTO heat_table_placements (packet_id, page, x, y, s) VALUES (%(packet)s, 1, 0.1, 0.2, 1)",
            "INSERT INTO packet_wh (packet_id, stage, by) VALUES (%(packet)s, 'incoming', %(user)s)",
            "INSERT INTO qc_inspections (packet_id, stage_id, sent_by, book, spool, location, signature_id) "
            "VALUES (%(packet)s, %(stage)s, %(user)s, true, true, 'Bay 2', %(sig)s)",
            "INSERT INTO packet_seals (packet_id, hash, key_id, sealed_by) "
            "VALUES (%(packet)s, '\\x01', 'abcd1234', %(user)s)",
            "INSERT INTO packet_visual_state (packet_id) VALUES (%(packet)s)",
            "INSERT INTO packet_docs (packet_id, asset_id, kind, title) VALUES (%(packet)s, %(asset)s, 'nde', 'RT')",
            "INSERT INTO qc_docs (unit_id, type, asset_id) VALUES (%(unit)s, 'WPS / PQR', %(asset)s)",
            "INSERT INTO itp_lines (itp_id, key, result, by) VALUES (%(itp)s, 'k1', 'approve', %(user)s)",
            "INSERT INTO itp_blocks (itp_id, block_key, signed_by, signature_id) "
            "VALUES (%(itp)s, 'b1', %(user)s, %(sig)s)",
        ]
        for stmt in statements:
            conn.execute(stmt, ids)
        conn.execute("UPDATE bom_lines SET pulled = true, row_version = row_version + 1 WHERE id = %(bom)s", ids)
        conn.execute(
            "UPDATE packets SET version = version + 1, next_weld_no = next_weld_no + 1 WHERE id = %(packet)s", ids
        )
        conn.execute("UPDATE packet_wh SET closed_at = now() WHERE packet_id = %(packet)s", ids)
        conn.execute("DELETE FROM co_items WHERE co_id = %(co)s", ids)
        wo = conn.execute("SELECT wo_base, lot, split, sub FROM packet_work_orders").fetchone()
        assert wo == ("0670001", "7A", "0", "0")
        for view in ("v_units", "v_open_cos"):
            conn.execute(q("SELECT * FROM {}", view)).fetchall()
            for priv in ("SELECT", "INSERT", "UPDATE", "DELETE"):
                allowed = one(conn, "SELECT has_table_privilege('ps_app', %s, %s)", (view, priv))
                assert allowed is (priv == "SELECT"), (view, priv)


FEED_TABLES = [
    "visual_batches",
    "visual_rows",
    "visual_materials",
    "visual_orders",
    "visual_order_links",
    "visual_labor",
    "visual_row_changes",
    "visual_material_changes",
    "visual_ignored_wos",
    "visual_resource_map",
    "drawing_lookups",
    "job_requests",
    "job_runs",
    "job_schedule",
    "worker_heartbeat",
]


def test_feed_tables_and_app_grants(pg_settings: Settings) -> None:
    with db.connect_as("migrate", pg_settings, autocommit=True) as conn:
        names = [r[0] for r in conn.execute("SELECT name FROM schema_migrations ORDER BY name").fetchall()]
    assert "0004_feed.sql" in names
    with app_conn(pg_settings) as conn:
        for table in FEED_TABLES:
            for priv in ("SELECT", "INSERT", "UPDATE", "DELETE"):
                assert one(conn, "SELECT has_table_privilege('ps_app', %s, %s)", (table, priv)) is True, (table, priv)
        jobs = [r[0] for r in conn.execute("SELECT job FROM job_schedule ORDER BY job").fetchall()]
        assert jobs == ["drawing-lookup", "housekeeping", "visual-feed"]
        assert one(conn, "SELECT target FROM visual_resource_map WHERE resource = 'WELD-PART'") == "03"
        assert one(conn, "SELECT target FROM visual_resource_map WHERE resource = 'Electical'") == "14"
        assert one(conn, "SELECT count(*) FROM visual_resource_map WHERE confirmed") == 0
        with pytest.raises(psycopg.errors.CheckViolation):
            conn.execute("INSERT INTO visual_resource_map (resource, target) VALUES ('X', 'nowhere')")


def test_feed_columns_are_text_ids(pg_settings: Settings) -> None:
    with app_conn(pg_settings) as conn:
        cols = dict(
            conn.execute(
                "SELECT table_name || '.' || column_name, data_type FROM information_schema.columns"
                " WHERE table_schema = 'public' AND column_name IN ('wo_base', 'lot', 'split', 'sub', 'seq', 'piece',"
                " 'part', 'cust_order_id', 'dwg')"
            ).fetchall()
        )
        for name, kind in cols.items():
            col = name.split(".")[1]
            if col in ("seq", "piece"):
                assert kind == "integer", name
            else:
                assert kind == "text", name
        for t, c in (
            ("packets", "visual_cancelled_at"),
            ("packets", "wo_gone_at"),
            ("packet_work_orders", "rev"),
            ("projects", "order_ref"),
            ("units", "customer_visual"),
            ("units", "order_pin"),
        ):
            assert (
                one(
                    conn,
                    "SELECT count(*) FROM information_schema.columns WHERE table_name = %s AND column_name = %s",
                    (t, c),
                )
                == 1
            ), (t, c)


def test_two_operations_of_one_work_order_on_one_packet(pg_settings: Settings) -> None:
    """Amendment 1: the key includes seq; hand rows (seq NULL) stay one per key."""
    with app_conn(pg_settings) as conn:
        ids = seed(conn)
        for seq in (10, 20):
            conn.execute(
                "INSERT INTO packet_work_orders (packet_id, wo_base, lot, split, sub, seq) "
                "VALUES (%s, 'L900-02', '1', '0', '0', %s)",
                (ids["packet"], seq),
            )
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute(
                "INSERT INTO packet_work_orders (packet_id, wo_base, lot, split, sub, seq) "
                "VALUES (%s, 'L900-02', '1', '0', '0', 20)",
                (ids["packet"],),
            )
        conn.execute(
            "INSERT INTO packet_work_orders (packet_id, wo_base, lot, split, sub) VALUES (%s, 'H1', '1', '0', '0')",
            (ids["packet"],),
        )
        with pytest.raises(psycopg.errors.UniqueViolation):
            conn.execute(
                "INSERT INTO packet_work_orders (packet_id, wo_base, lot, split, sub) VALUES (%s, 'H1', '1', '0', '0')",
                (ids["packet"],),
            )


def test_notification_events(pg_settings: Settings) -> None:
    """0005: the notification placeholder. One open row per kind and subject; the app may write it."""
    with db.connect_as("migrate", pg_settings, autocommit=True) as conn:
        names = [r[0] for r in conn.execute("SELECT name FROM schema_migrations ORDER BY name").fetchall()]
    assert "0005_warehouse.sql" in names
    with app_conn(pg_settings) as conn:
        for priv in ("SELECT", "INSERT", "UPDATE", "DELETE"):
            assert one(conn, "SELECT has_table_privilege('ps_app', 'notification_events', %s)", (priv,)) is True
        ids = seed(conn)
        first = one(
            conn,
            "INSERT INTO notification_events (kind, subject, detail) VALUES ('t_kind', 'assets', '{\"pct\": 90}')"
            " RETURNING id",
        )
        assert one(conn, "SELECT severity FROM notification_events WHERE id = %s", (first,)) == "warning"
        with pytest.raises(psycopg.errors.UniqueViolation), conn.transaction():
            conn.execute("INSERT INTO notification_events (kind, subject) VALUES ('t_kind', 'assets')")
        with pytest.raises(psycopg.errors.CheckViolation), conn.transaction():
            conn.execute("INSERT INTO notification_events (kind, severity) VALUES ('t_other', 'loud')")
        # A null subject is one subject: a second open row for it is refused too.
        conn.execute("INSERT INTO notification_events (kind) VALUES ('t_null')")
        with pytest.raises(psycopg.errors.UniqueViolation), conn.transaction():
            conn.execute("INSERT INTO notification_events (kind) VALUES ('t_null')")
        conn.execute(
            "UPDATE notification_events SET resolved_at = now(), acknowledged_at = now(), acknowledged_by = %s"
            " WHERE id = %s",
            (ids["user"], first),
        )
        conn.execute("INSERT INTO notification_events (kind, subject) VALUES ('t_kind', 'assets')")
        assert one(conn, "SELECT count(*) FROM notification_events WHERE kind = 't_kind'") == 2


def test_shop_columns_and_checklist_templates(pg_settings: Settings) -> None:
    """0006: label positions on welds and heat tags, carried_from on heat table placements, the checklist row."""
    with db.connect_as("migrate", pg_settings, autocommit=True) as conn:
        names = [r[0] for r in conn.execute("SELECT name FROM schema_migrations ORDER BY name").fetchall()]
    assert "0006_shop.sql" in names
    with app_conn(pg_settings) as conn:
        for t, c, kind in (
            ("welds", "bfx", "double precision"),
            ("welds", "bfy", "double precision"),
            ("heat_tags", "bfx", "double precision"),
            ("heat_tags", "bfy", "double precision"),
            ("heat_table_placements", "carried_from", "integer"),
        ):
            got = one(
                conn,
                "SELECT data_type FROM information_schema.columns WHERE table_name = %s AND column_name = %s",
                (t, c),
            )
            assert got == kind, (t, c)
        ids = seed(conn)
        with pytest.raises(psycopg.errors.CheckViolation), conn.transaction():
            conn.execute(
                "INSERT INTO welds (packet_id, weld_no, items, bfx) VALUES (%s, 1, '{1}', 1.5)", (ids["packet"],)
            )
        value = one(conn, "SELECT value FROM settings WHERE key = 'checklist_templates'")
        assert [i["key"] for i in value["weld"]] == ["dwg", "heat", "wps", "fit", "pre", "vt", "stamp", "clean", "qty"]
        assert [i["key"] for i in value["plain"]] == ["dwg", "prev", "mat", "work", "insp", "qty"]
        assert value["weld"][0]["text"] == "Drawing {drawing} Rev {rev} matches the packet"


def test_qc_columns_checks_and_items(pg_settings: Settings) -> None:
    """0007: the inspection NCR, document notes and kinds, Q.C. folder scope and types, the ITP details, the ITP log
    and the Q.C. check items row."""
    with db.connect_as("migrate", pg_settings, autocommit=True) as conn:
        names = [r[0] for r in conn.execute("SELECT name FROM schema_migrations ORDER BY name").fetchall()]
    assert "0007_qc.sql" in names
    with app_conn(pg_settings) as conn:
        for t, c, kind in (
            ("qc_inspections", "ncr", "text"),
            ("packet_docs", "note", "text"),
            ("qc_docs", "title", "text"),
            ("qc_docs", "note", "text"),
            ("qc_docs", "superseded_at", "timestamp with time zone"),
            ("qc_docs", "superseded_by_user", "uuid"),
            ("itps", "no", "text"),
            ("itps", "hdr", "jsonb"),
            ("itps", "created_at", "timestamp with time zone"),
            ("itps", "created_by", "uuid"),
            ("itp_lines", "initials", "text"),
            ("itp_lines", "cust_code", "text"),
            ("itp_lines", "cust_by", "text"),
            ("itp_lines", "cust_at", "date"),
            ("itp_lines", "cust_rec", "uuid"),
            ("itp_lines", "cust_rec_at", "timestamp with time zone"),
            ("itp_rejects", "kind", "text"),
            ("itp_blocks", "qn", "text"),
            ("itp_blocks", "remarks", "text"),
            ("itp_blocks", "cust_name", "text"),
            ("itp_blocks", "cust_at", "date"),
            ("itp_blocks", "cust_by", "uuid"),
            ("itp_blocks", "cust_rec_at", "timestamp with time zone"),
            ("itp_blocks", "row_version", "integer"),
            ("itp_log", "it_reason", "text"),
        ):
            got = one(
                conn,
                "SELECT data_type FROM information_schema.columns WHERE table_name = %s AND column_name = %s",
                (t, c),
            )
            assert got == kind, (t, c)
        ids = seed(conn)
        ids["asset"] = one(
            conn,
            "INSERT INTO assets (sha256, size, mime, storage_key, pages) VALUES ('\\x00', 10, 'application/pdf',"
            " 'ab/q7', 1) RETURNING id",
        )
        with pytest.raises(psycopg.errors.CheckViolation), conn.transaction():
            conn.execute(
                "INSERT INTO packet_docs (packet_id, asset_id, kind) VALUES (%(packet)s, %(asset)s, 'mtr')", ids
            )
        with pytest.raises(psycopg.errors.CheckViolation), conn.transaction():
            conn.execute(
                "INSERT INTO qc_docs (project_id, unit_id, type, asset_id) VALUES (%(project)s, %(unit)s, 'ITP',"
                " %(asset)s)",
                ids,
            )
        with pytest.raises(psycopg.errors.CheckViolation), conn.transaction():
            conn.execute("INSERT INTO qc_docs (unit_id, type, asset_id) VALUES (%(unit)s, 'wps', %(asset)s)", ids)
        conn.execute("INSERT INTO qc_docs (project_id, type, asset_id) VALUES (%(project)s, 'ITP', %(asset)s)", ids)
        # A reject may come before its NCR number; a block row exists before it is signed.
        conn.execute("INSERT INTO itp_rejects (itp_id, line_key) VALUES (%(itp)s, '1.1')", ids)
        assert one(conn, "SELECT kind FROM itp_rejects WHERE itp_id = %(itp)s", ids) == "reject"
        conn.execute("INSERT INTO itp_blocks (itp_id, block_key) VALUES (%(itp)s, '3')", ids)
        assert one(conn, "SELECT signed_at FROM itp_blocks WHERE itp_id = %(itp)s", ids) is None
        conn.execute("UPDATE itps SET status = 'complete' WHERE id = %(itp)s", ids)
        value = one(conn, "SELECT value FROM settings WHERE key = 'qc_check_items'")
        assert [g["group"] for g in value["groups"]] == ["Data book package", "Physical spool"]
        keys = [i["key"] for g in value["groups"] for i in g["items"]]
        assert keys == [
            "db_heats",
            "db_mtr",
            "db_weldmap",
            "db_wps",
            "db_ck",
            "ph_dim",
            "ph_heat",
            "ph_stamp",
            "ph_clean",
            "ph_qty",
        ]