spencermandClaude Fable 5.1
00531cb329
The Q.C. screen (KPIs, boards, project folders), the Q.C. tab on the packet page (check items from the qc_check_items settings row, visual exam per weld, NDE, hardness with the NACE label, notes), accept with a route and a signature gated by qc_missing_empty, reject with a reason or an NCR number closing the inspection into qc_history, Q.C. folder documents per project and unit with superseding revisions, packet documents as a closed set with take-out reasons, the digital ITP per unit from the frozen template (line initials with CAS, unsign, N/A, customer codes, rejects with NCR, block signatures, sign-off with the PDF hook, admin reopen, the ITP log), hold-point warnings by department page group, the scoped policy check for unit and project writes, migration 0007, tests, Playwright scenarios and docs. Co-Authored-By: Claude Fable 5.1 <[email protected]>
387 lines
16 KiB
Python
387 lines
16 KiB
Python
"""Phase 6 Q.C. inspection API: the check and weld PUTs, the field CAS PATCH, the qc_items hints, nace_label, accept
|
|
to the data book, to shipping and to a department, reject with a reason, an NCR or both, the second inspection
|
|
after a reject, and the Q.C. home board. Every log text is the contract's, exactly."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fakedir import FakeDirectory
|
|
from pkthelpers import step
|
|
from pshelpers import needs_pg
|
|
from qchelpers import (
|
|
NCR,
|
|
NCR_URL,
|
|
QC_KEYS,
|
|
ROUTE_DB,
|
|
accept,
|
|
accept_route,
|
|
at_qc,
|
|
finish_inspection,
|
|
qc_check,
|
|
qc_patch,
|
|
qc_weld,
|
|
reject,
|
|
to_qc,
|
|
)
|
|
from shophelpers import (
|
|
SIGN,
|
|
Shop,
|
|
all_rows,
|
|
fresh,
|
|
issue_co,
|
|
last_log,
|
|
log_rows,
|
|
make_unit,
|
|
ok,
|
|
one,
|
|
open_shop,
|
|
refused,
|
|
run,
|
|
set_stamp,
|
|
)
|
|
|
|
from packetstation.settings import Settings
|
|
|
|
pytestmark = needs_pg
|
|
|
|
REASON = "Weld 2 undercut too deep"
|
|
ITEM_TEXT = {
|
|
"db_heats": "Heat number recorded for every BOM item",
|
|
"db_mtr": "MTRs on file and match the heat numbers",
|
|
"db_weldmap": "Weld map complete: every weld identified",
|
|
"db_wps": "Welds made to WPS-11; welder [75] qualified",
|
|
"db_ck": "Stall checklist and weld record complete",
|
|
"ph_dim": "Dimensions verified against D-5001 Rev B",
|
|
"ph_heat": "Heat numbers on material match the heat map",
|
|
"ph_stamp": "Welds stamped with welder ID",
|
|
"ph_clean": "Clean: no slag, spatter or arc strikes",
|
|
"ph_qty": "Quantity matches the work orders",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
|
|
yield from open_shop(pg_settings, fake_dir)
|
|
|
|
|
|
def actions(shop: Shop, pid: int) -> list[str]:
|
|
return [r["action"] for r in log_rows(shop("qc.quinn"), pid)]
|
|
|
|
|
|
# ================================================================ checks, welds, fields
|
|
|
|
|
|
def test_check_put_per_key(shop: Shop) -> None:
|
|
set_stamp(shop.settings, "weld.andy", "75")
|
|
pkt = at_qc(shop)
|
|
q = shop("qc.quinn")
|
|
pid = pkt["id"]
|
|
ok(qc_patch(q, pid, {"stamp": "Q1"}))
|
|
# the item texts carry the drawing, the rev, the WPS and the welder stamp
|
|
texts = {i["key"]: i["text"] for i in fresh(q, pid)["qc_items"]}
|
|
assert texts == ITEM_TEXT
|
|
for key in QC_KEYS:
|
|
before = len(log_rows(q, pid))
|
|
out = ok(qc_check(q, pid, key))
|
|
entry = out["qc"]["checks"][key]
|
|
assert entry["done"] is True and entry["at"] and entry["by"], entry
|
|
assert entry["stamp"] == "Q1" and entry["by_name"] == "Quinn QC", entry
|
|
log = last_log(q, pid)
|
|
assert log["action"] == f"Q.C. checked: {ITEM_TEXT[key]}" and not log["note"], log
|
|
assert len(log_rows(q, pid)) == before + 1
|
|
# a tick that changes nothing writes nothing
|
|
before = len(log_rows(q, pid))
|
|
ok(qc_check(q, pid, "ph_dim"))
|
|
assert len(log_rows(q, pid)) == before
|
|
# untick
|
|
out = ok(qc_check(q, pid, "ph_dim", done=False))
|
|
assert out["qc"]["checks"]["ph_dim"].get("done") is False
|
|
assert last_log(q, pid)["action"] == "Q.C. unchecked: Dimensions verified against D-5001 Rev B"
|
|
# an unknown key
|
|
refused(qc_check(q, pid, "ph_paint"), 422, "invalid_key")
|
|
# Q.C. only
|
|
refused(qc_check(shop("sup.wes"), pid, "ph_dim"), 403, "forbidden")
|
|
|
|
|
|
def test_check_put_without_inspection(shop: Shop) -> None:
|
|
pkt = at_qc(shop)
|
|
out = ok(reject(shop, pkt, REASON))
|
|
assert out["qc"] is None
|
|
# back with the supervisor: no inspection is open
|
|
r = qc_check(shop("qc.quinn"), pkt["id"], "ph_dim")
|
|
assert r.status_code in (403, 409), r.text
|
|
assert r.json()["code"] in ("wrong_status", "no_inspection"), r.text
|
|
|
|
|
|
def test_weld_put(shop: Shop) -> None:
|
|
pkt = at_qc(shop)
|
|
q = shop("qc.quinn")
|
|
pid = pkt["id"]
|
|
w1, w2 = sorted(int(w["weld_no"]) for w in pkt["welds"])
|
|
out = ok(qc_weld(q, pid, w1, result="accept"))
|
|
e = out["qc"]["weld_exam"][str(w1)]
|
|
assert e["result"] == "accept" and e["at"] and e["by"] and e["by_name"] == "Quinn QC", e
|
|
assert last_log(q, pid)["action"] == f"VT accept · W{w1}" and not last_log(q, pid)["note"]
|
|
out = ok(qc_weld(q, pid, w2, result="reject", note="undercut"))
|
|
assert out["qc"]["weld_exam"][str(w2)]["note"] == "undercut"
|
|
assert (last_log(q, pid)["action"], last_log(q, pid)["note"]) == (f"VT reject · W{w2}", "undercut")
|
|
# a note-only change
|
|
ok(qc_weld(q, pid, w2, note="undercut 1/32"))
|
|
assert (last_log(q, pid)["action"], last_log(q, pid)["note"]) == ("Q.C. record changed", f"W{w2} note")
|
|
# cleared
|
|
out = ok(qc_weld(q, pid, w2, result=""))
|
|
assert not out["qc"]["weld_exam"].get(str(w2), {}).get("result")
|
|
assert last_log(q, pid)["action"] == f"VT cleared · W{w2}"
|
|
# a weld that is not on the packet
|
|
refused(qc_weld(q, pid, 99, result="accept"), 422, "invalid_weld")
|
|
refused(qc_weld(shop("clerk.clara"), pid, w1, result="reject"), 403, "forbidden")
|
|
|
|
|
|
def test_field_cas_patch(shop: Shop) -> None:
|
|
pkt = at_qc(shop)
|
|
q = shop("qc.quinn")
|
|
pid = pkt["id"]
|
|
out = ok(qc_patch(q, pid, {"inspector": "Quinn QC", "stamp": "Q1"}, {"inspector": None, "stamp": None}))
|
|
assert out["qc"]["inspector"] == "Quinn QC" and out["qc"]["stamp"] == "Q1"
|
|
assert (last_log(q, pid)["action"], last_log(q, pid)["note"]) == (
|
|
"Q.C. record changed",
|
|
"inspector Quinn QC · stamp Q1",
|
|
)
|
|
# a stale base
|
|
r = refused(qc_patch(q, pid, {"inspector": "Q. Quinn"}, {"inspector": "Somebody"}), 409, "field_conflict")
|
|
assert r["current"] == {"inspector": "Quinn QC"}
|
|
# NDE method and result validation
|
|
r = refused(qc_patch(q, pid, {"nde_method": "XR"}), 422, "invalid_field")
|
|
assert r["detail"] == "Pick an NDE method from the list."
|
|
refused(qc_patch(q, pid, {"nde_result": "Maybe"}), 422, "invalid_field")
|
|
ok(qc_patch(q, pid, {"nde_method": "RT"}))
|
|
assert last_log(q, pid)["action"] == "NDE method: RT"
|
|
out = ok(qc_patch(q, pid, {"nde_result": "Acceptable"}))
|
|
assert out["qc"]["nde_result"] == "Acceptable"
|
|
assert last_log(q, pid)["action"] == "NDE result: Acceptable"
|
|
out = ok(qc_patch(q, pid, {"hardness": "198 HB", "nde_report": "RPT-1"}))
|
|
assert (last_log(q, pid)["action"], last_log(q, pid)["note"]) == (
|
|
"Q.C. record changed",
|
|
"hardness 198 HB · report RPT-1",
|
|
)
|
|
ok(qc_patch(q, pid, {"notes": "Fit-up good"}))
|
|
assert (last_log(q, pid)["action"], last_log(q, pid)["note"]) == ("Q.C. record changed", "notes changed")
|
|
ok(qc_patch(q, pid, {"nde_result": ""}))
|
|
assert last_log(q, pid)["action"] == "NDE result: -"
|
|
ok(qc_patch(q, pid, {"nde_result": "Acceptable"}))
|
|
# None required clears the result and the report
|
|
out = ok(qc_patch(q, pid, {"nde_method": "None required"}))
|
|
assert out["qc"]["nde_method"] == "None required"
|
|
assert not out["qc"]["nde_result"] and not out["qc"]["nde_report"]
|
|
assert "NDE method: None required" in actions(shop, pid)
|
|
for method in ("UT", "MT", "PT"):
|
|
assert ok(qc_patch(q, pid, {"nde_method": method}))["qc"]["nde_method"] == method
|
|
# the old flat body with If-Match still merges checks (phase 2), or the new endpoints replaced it
|
|
cur = fresh(q, pid)["qc"]
|
|
r = q.patch(
|
|
f"/api/packets/{pid}/qc",
|
|
json={"checks": {"ph_qty": {"done": True}}},
|
|
headers={"If-Match": str(cur["row_version"])},
|
|
)
|
|
assert r.status_code < 500, r.text
|
|
# the nace field is computed, not written
|
|
r = qc_patch(q, pid, {"nace": True})
|
|
assert r.status_code < 500, r.text
|
|
assert fresh(q, pid)["qc"]["nace_label"] is False
|
|
|
|
|
|
def test_qc_items_hints_and_nace_label(shop: Shop) -> None:
|
|
pkt = at_qc(shop)
|
|
q = shop("qc.quinn")
|
|
items = fresh(q, pkt["id"])["qc_items"]
|
|
assert [i["key"] for i in items] == list(QC_KEYS)
|
|
assert [i["group"] for i in items] == ["Data book package"] * 5 + ["Physical spool"] * 5
|
|
hints = {i["key"]: i["hint"] for i in items}
|
|
assert hints["db_heats"] == "3/3 recorded"
|
|
assert hints["db_mtr"] == "0/3 MTRs attached"
|
|
assert hints["db_weldmap"] == "2 welds on map"
|
|
assert hints["db_ck"] == "checklist 9/9 · filler ER70S-6"
|
|
assert all(hints[k] is None for k in QC_KEYS[5:]), hints
|
|
assert fresh(q, pkt["id"])["qc"]["nace_label"] is False
|
|
run(shop.settings, "UPDATE packets SET wps = 'WPS-12 nace sour' WHERE id = %s", (pkt["id"],))
|
|
assert fresh(q, pkt["id"])["qc"]["nace_label"] is True
|
|
|
|
|
|
# ================================================================ accept
|
|
|
|
|
|
def _snapshot(shop: Shop, pid: int) -> Any:
|
|
row = one(
|
|
shop.settings,
|
|
"SELECT qc_snapshot FROM packet_stages WHERE packet_id = %s AND closed_at IS NOT NULL ORDER BY id DESC LIMIT 1",
|
|
(pid,),
|
|
)
|
|
assert row is not None
|
|
return row[0]
|
|
|
|
|
|
def test_accept_to_data_book(shop: Shop) -> None:
|
|
pkt = finish_inspection(shop, at_qc(shop), nde=("RT", "Acceptable"))
|
|
out = ok(accept(shop, pkt, "databook"))
|
|
assert out["status"] == "complete" and out["routed_to"] == "databook" and out["qc"] is None
|
|
assert out["stage"] is None
|
|
h = out["qc_history"][-1]
|
|
assert h["result"] == "accepted" and h["inspector"] == "Quinn QC"
|
|
assert h["by"]["display_name"] == "Quinn QC" and h["location"] == "Bay 2" and h["sent_at"]
|
|
assert h["rejected_welds"] == []
|
|
assert out["qc_last"]["id"] == h["id"] and out["qc_last"]["result"] == "accepted"
|
|
snap = _snapshot(shop, pkt["id"])
|
|
assert snap and snap.get("inspector") == "Quinn QC" and snap.get("result") == "accepted", snap
|
|
log = last_log(shop("qc.quinn"), pkt["id"])
|
|
assert log["action"] == f"Q.C. accepted → {ROUTE_DB}"
|
|
assert log["note"] == "Quinn QC [Q1] · 2 welds VT accepted · RT Acceptable"
|
|
assert log["signed"] is True
|
|
|
|
|
|
def test_accept_to_shipping(shop: Shop) -> None:
|
|
pkt = finish_inspection(shop, at_qc(shop))
|
|
out = ok(accept(shop, pkt, "shipping"))
|
|
assert out["status"] == "complete" and out["routed_to"] == "shipping"
|
|
log = last_log(shop("qc.quinn"), pkt["id"])
|
|
assert (log["action"], log["note"]) == ("Q.C. accepted → Shipping", "Quinn QC [Q1] · 2 welds VT accepted")
|
|
|
|
|
|
def test_accept_note_without_stamp(shop: Shop) -> None:
|
|
pkt = finish_inspection(shop, at_qc(shop, welds=1), stamp="")
|
|
ok(accept(shop, pkt, "shipping"))
|
|
log = last_log(shop("qc.quinn"), pkt["id"])
|
|
assert log["note"] == "Quinn QC · 1 weld VT accepted" or log["note"] == "Quinn QC · 1 welds VT accepted", log
|
|
|
|
|
|
def test_accept_and_route_to_hydro(shop: Shop) -> None:
|
|
pkt = finish_inspection(shop, at_qc(shop))
|
|
refused(accept_route(shop, pkt, "03"), 403, "precondition")
|
|
out = ok(accept_route(shop, pkt, "09"))
|
|
assert out["status"] == "ready" and out["dept"]["code"] == "09"
|
|
assert out["stage"] is not None and out["stage"]["dept_code"] == "09"
|
|
assert out["qc"] is None and out["qc_last"]["result"] == "accepted"
|
|
log = last_log(shop("qc.quinn"), pkt["id"])
|
|
assert (log["action"], log["note"]) == ("Q.C. accepted → 09 Hydro", "Quinn QC [Q1] · from 03 Weld Parts")
|
|
|
|
|
|
# ================================================================ reject
|
|
|
|
|
|
def _rejected(shop: Shop, pkt: dict[str, Any], out: dict[str, Any], note: str, return_note: str) -> None:
|
|
assert out["status"] == "review" and out["qc"] is None
|
|
last = out["qc_last"]
|
|
assert last["result"] == "rejected"
|
|
assert out["return_note"] == return_note
|
|
log = last_log(shop("sup.wes"), pkt["id"])
|
|
assert (log["action"], log["note"]) == ("Q.C. rejected → supervisor", note)
|
|
assert log["signed"] is True
|
|
|
|
|
|
def test_reject_with_reason_only(shop: Shop) -> None:
|
|
pkt = at_qc(shop)
|
|
w2 = max(int(w["weld_no"]) for w in pkt["welds"])
|
|
ok(qc_weld(shop("qc.quinn"), pkt["id"], w2, result="reject", note="undercut"))
|
|
out = ok(reject(shop, pkt, REASON))
|
|
_rejected(shop, pkt, out, f"{REASON} · welds W{w2}", REASON)
|
|
assert out["qc_last"]["ncr"] is None and out["qc_last"]["rejected_welds"] == [f"W{w2}"]
|
|
|
|
|
|
def test_reject_with_ncr_only(shop: Shop) -> None:
|
|
pkt = at_qc(shop)
|
|
q = shop("qc.quinn")
|
|
refused(reject(shop, pkt, ncr="NCR-26-1"), 422, "invalid_ncr", "Enter an NCR number like NCR-2026-0021.")
|
|
refused(reject(shop, pkt), 400, "reason_required")
|
|
refused(reject(shop, pkt, "bad"), 400, "reason_required")
|
|
assert fresh(q, pkt["id"])["status"] == "qc"
|
|
out = ok(reject(shop, pkt, ncr=" ncr-2026-0021 "))
|
|
_rejected(shop, pkt, out, NCR, NCR)
|
|
assert out["qc_last"]["ncr"] == NCR and out["qc_last"]["rejected_welds"] == []
|
|
row = one(
|
|
shop.settings, "SELECT ncr FROM qc_inspections WHERE packet_id = %s ORDER BY id DESC LIMIT 1", (pkt["id"],)
|
|
)
|
|
assert row[0] == NCR
|
|
assert NCR_URL.endswith("/ncrs/")
|
|
|
|
|
|
def test_reject_with_both_then_send_again(shop: Shop) -> None:
|
|
pkt = at_qc(shop)
|
|
q = shop("qc.quinn")
|
|
w1, w2 = sorted(int(w["weld_no"]) for w in pkt["welds"])
|
|
ok(qc_check(q, pkt["id"], "ph_dim"))
|
|
ok(qc_weld(q, pkt["id"], w2, result="reject", note="porosity"))
|
|
ok(qc_weld(q, pkt["id"], w1, result="reject", note="undercut"))
|
|
out = ok(reject(shop, pkt, "Weld 2 undercut", ncr=NCR))
|
|
_rejected(shop, pkt, out, f"{NCR} · Weld 2 undercut · welds W{w1}, W{w2}", f"{NCR} · Weld 2 undercut")
|
|
first = out["qc_last"]
|
|
assert first["ncr"] == NCR and first["rejected_welds"] == [f"W{w1}", f"W{w2}"]
|
|
assert first["by"]["display_name"] == "Quinn QC" and first["location"] == "Bay 2"
|
|
assert len(out["qc_history"]) == 1
|
|
# the supervisor sees it and sends it to Q.C. again: a fresh empty inspection
|
|
wes = fresh(shop("sup.wes"), pkt["id"])
|
|
assert wes["qc_last"]["result"] == "rejected"
|
|
again = to_qc(shop, wes, location="Bay 4")
|
|
assert again["qc"]["checks"] == {} and again["qc"]["weld_exam"] == {}
|
|
assert again["qc"]["location"] == "Bay 4" and again["qc"]["ncr"] is None
|
|
assert len(again["qc_history"]) == 1 and again["qc_history"][0]["ncr"] == NCR
|
|
assert again["qc_last"]["id"] == first["id"]
|
|
rows = all_rows(shop.settings, "SELECT count(*) FROM qc_inspections WHERE packet_id = %s", (pkt["id"],))
|
|
assert rows[0][0] == 2
|
|
|
|
|
|
def test_co_hold_blocks_accept_but_not_reject(shop: Shop) -> None:
|
|
pkt = finish_inspection(shop, at_qc(shop))
|
|
pkt = issue_co(shop, pkt, ["03"])
|
|
refused(accept(shop, pkt), 403, "co_hold")
|
|
out = ok(reject(shop, fresh(shop("qc.quinn"), pkt["id"]), REASON))
|
|
assert out["status"] == "review"
|
|
|
|
|
|
# ================================================================ the Q.C. home
|
|
|
|
|
|
def _ids(cards: list[dict[str, Any]]) -> list[int]:
|
|
return [int(c["id"]) for c in cards]
|
|
|
|
|
|
def test_qc_home(shop: Shop) -> None:
|
|
q = shop("qc.quinn")
|
|
_, unit_a = make_unit(shop.settings)
|
|
_, unit_b = make_unit(shop.settings)
|
|
waiting = at_qc(shop, unit_a)
|
|
to_reject = at_qc(shop, unit_b)
|
|
home = ok(q.get("/api/qc/home"))
|
|
assert set(home) >= {"kpis", "columns", "folders"}
|
|
assert home["kpis"] == {"waiting": 2, "accepted_7d": 0, "rejected_7d": 0}
|
|
assert _ids(home["columns"]["ready"]) == [waiting["id"], to_reject["id"]] # oldest status_at first
|
|
card = home["columns"]["ready"][0]
|
|
assert card["qc"]["location"] == "Bay 2" and card["qc"]["sent_at"]
|
|
assert "itp_hold_count" in card
|
|
assert home["columns"]["accepted"] == [] and home["columns"]["rejected"] == []
|
|
|
|
ok(reject(shop, to_reject, ncr=NCR))
|
|
accepted = ok(accept(shop, finish_inspection(shop, waiting), "databook"))
|
|
home = ok(q.get("/api/qc/home"))
|
|
assert home["kpis"] == {"waiting": 0, "accepted_7d": 1, "rejected_7d": 1}
|
|
assert _ids(home["columns"]["accepted"]) == [accepted["id"]]
|
|
assert _ids(home["columns"]["rejected"]) == [to_reject["id"]]
|
|
assert home["columns"]["rejected"][0]["qc"]["result"] == "rejected"
|
|
assert home["columns"]["accepted"][0]["qc"]["result"] == "accepted"
|
|
|
|
# the unit filter narrows the request
|
|
narrowed = ok(q.get(f"/api/qc/home?unit_id={unit_b}"))
|
|
assert _ids(narrowed["columns"]["rejected"]) == [to_reject["id"]] and narrowed["columns"]["accepted"] == []
|
|
|
|
# pm.paula reads it; a session is needed
|
|
assert ok(shop("pm.paula").get("/api/qc/home"))["kpis"]["accepted_7d"] == 1
|
|
assert ok(shop("wh.wade").get("/api/qc/home"))["kpis"]["rejected_7d"] == 1
|
|
|
|
|
|
def test_transition_stale_expect(shop: Shop) -> None:
|
|
pkt = finish_inspection(shop, at_qc(shop))
|
|
ok(reject(shop, pkt, REASON))
|
|
r = refused(step(shop("qc.quinn"), pkt, "qc_accept", routed_to="databook", sign=SIGN), 409, "stale_step")
|
|
assert r["current"]["status"] == "review"
|