spencermandClaude Fable 5.1
5a4174e0c2
The shop at parity with the old build. Supervisor screen (department chips, CO headline, numbers, Change orders
on hold, station grid with clocks and CO tags, boards) and Stall screen (station picker, hero, My packets,
Submitted). Shop steps on the packet page: assign with station, welder picker and stamp; pull back; Start, Pause,
Resume; Submit with a PIN signature and the missing steps as links; review with Send to Q.C., Return to station
and Skip Q.C. route; a lead's read-only review. Weld map (pdf.js viewer with pan, pinch, zoom, full screen; marks
with leader lines; weld sheet; weld record with field CAS; carried marks). Heat map (heat table placed, dragged,
sized, removed; legacy amber tags; borrowed heat map read-only). Checklist tab with templates from settings.
VOID stamps on voided pages. Change orders: CO-n labels, supervisor verify with signature, approval copies the
heat table to the new sheet and the BOM merge keeps the shop's values. Server clock of record with VISUAL ticket
hours beside it. Presence per packet over SSE. Migration 0006 (bfx, bfy, carried_from, checklist_templates).
New endpoints /api/shop/board, /api/shop/stations, /api/shop/stall, /api/departments/{code}/people.
Tests: policy oracle for every shop action, role and status; clock arithmetic; concurrent inserts; CO apply;
frontend geometry, heat table, clock, checklist, weld sheet, screens; seven Playwright scenarios. Docs updated.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
530 lines
25 KiB
Python
530 lines
25 KiB
Python
"""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"],
|
|
}
|
|
|
|
# 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')",
|
|
"co_verifications": (
|
|
"INSERT INTO co_verifications (co_id, dept_code, by, signature_id) VALUES (%(co)s, 'WH', %(user)s, %(sig)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, 'mtr', 'MTR')",
|
|
"INSERT INTO qc_docs (project_id, unit_id, type, asset_id) "
|
|
"VALUES (%(project)s, %(unit)s, 'wps', %(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"
|