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]>
335 lines
16 KiB
Python
335 lines
16 KiB
Python
"""Phase 6 documents: packet documents for the data book (add, the closed kind set, the file type, the filed lock,
|
|
take out with a reason and the row kept) and Q.C. folder documents on a unit or a project (add, the closed type
|
|
set, who may add, new revisions that supersede and never delete, the folder reads and counts, the audit texts)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fakedir import FakeDirectory
|
|
from pshelpers import needs_pg
|
|
from qchelpers import (
|
|
DOC_KINDS,
|
|
MSG_DOC_KIND,
|
|
MSG_TAKE_OUT,
|
|
QC_TYPES,
|
|
add_packet_doc,
|
|
csv_asset,
|
|
filed_packet,
|
|
folder_project,
|
|
folder_unit,
|
|
make_qc_doc,
|
|
start_itp,
|
|
take_out,
|
|
)
|
|
from shophelpers import (
|
|
IT,
|
|
Shop,
|
|
all_rows,
|
|
draft,
|
|
fresh,
|
|
last_log,
|
|
make_unit,
|
|
ok,
|
|
one,
|
|
open_shop,
|
|
refused,
|
|
run,
|
|
)
|
|
|
|
from packetstation.settings import Settings
|
|
|
|
pytestmark = needs_pg
|
|
|
|
|
|
@pytest.fixture
|
|
def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
|
|
yield from open_shop(pg_settings, fake_dir)
|
|
|
|
|
|
def numbers(shop: Shop, unit_id: int) -> tuple[str, str]:
|
|
"""(project number, unit number) of a unit."""
|
|
row = one(
|
|
shop.settings,
|
|
"SELECT p.number, p.number || u.lot FROM units u JOIN projects p ON p.id = u.project_id WHERE u.id = %s",
|
|
(unit_id,),
|
|
)
|
|
return str(row[0]), str(row[1])
|
|
|
|
|
|
def add_unit(shop: Shop, project_id: int, lot: str) -> int:
|
|
row = one(
|
|
shop.settings,
|
|
"INSERT INTO units (project_id, lot, name, status) VALUES (%s, %s, 'Unit two', 'open') RETURNING id",
|
|
(project_id, lot),
|
|
)
|
|
return int(row[0])
|
|
|
|
|
|
def doc_of(pkt: dict[str, Any], doc_id: int) -> dict[str, Any]:
|
|
return next(d for d in pkt["docs"] if int(d["id"]) == doc_id)
|
|
|
|
|
|
# ================================================================ packet documents
|
|
|
|
|
|
def test_packet_doc_add(shop: Shop) -> None:
|
|
pkt = draft(shop)
|
|
q = shop("qc.quinn")
|
|
out = ok(add_packet_doc(q, pkt["id"], "nde", "RT report W1-W4", "RPT-2210, accept", "rt.pdf"))
|
|
d = out["docs"][-1]
|
|
assert d["kind"] == "nde" and d["kind_label"] == DOC_KINDS["nde"]
|
|
assert d["title"] == "RT report W1-W4" and d["note"] == "RPT-2210, accept"
|
|
assert d["file_name"] == "rt.pdf" and d["size"] > 0 and d["removed"] is None
|
|
log = last_log(q, pkt["id"])
|
|
assert (log["action"], log["note"]) == (
|
|
"Document added",
|
|
"NDE report (RT / UT / MT / PT) · RT report W1-W4 · rt.pdf",
|
|
)
|
|
# clerk.clara adds one with no title: the title is the kind label
|
|
out = ok(add_packet_doc(shop("clerk.clara"), pkt["id"], "hydro", None, None, "hydro.pdf"))
|
|
d = out["docs"][-1]
|
|
assert d["title"] == "Hydrotest report" and d["kind_label"] == "Hydrotest report"
|
|
assert last_log(q, pkt["id"])["note"] == "Hydrotest report · Hydrotest report · hydro.pdf"
|
|
# every kind of the closed set is accepted
|
|
for kind, label in DOC_KINDS.items():
|
|
out = ok(add_packet_doc(q, pkt["id"], kind, f"Doc {kind}"))
|
|
assert out["docs"][-1]["kind_label"] == label
|
|
assert [d["kind"] for d in out["docs"]][:2] == ["nde", "hydro"] # ordered added_at, id
|
|
|
|
|
|
def test_packet_doc_refusals(shop: Shop) -> None:
|
|
pkt = draft(shop)
|
|
q = shop("qc.quinn")
|
|
refused(add_packet_doc(q, pkt["id"], "xray"), 422, "invalid_kind", MSG_DOC_KIND)
|
|
refused(add_packet_doc(q, pkt["id"], asset_id=csv_asset(q)["id"]), 422, "invalid_asset")
|
|
refused(add_packet_doc(shop("wh.wade"), pkt["id"]), 403, "forbidden")
|
|
refused(add_packet_doc(shop("sup.wes"), pkt["id"]), 403, "forbidden")
|
|
assert fresh(q, pkt["id"])["docs"] == []
|
|
# over the note limit: refused, or cut to the limit (the contract gives the limit, not the answer)
|
|
r = add_packet_doc(q, pkt["id"], note="x" * 301)
|
|
assert r.status_code == 422 or len(r.json()["docs"][-1]["note"]) <= 300, r.text
|
|
|
|
|
|
def test_packet_doc_note_limit(shop: Shop) -> None:
|
|
pkt = draft(shop)
|
|
r = add_packet_doc(shop("qc.quinn"), pkt["id"], note="x" * 300)
|
|
if r.status_code == 200:
|
|
assert len(r.json()["docs"][-1]["note"]) == 300
|
|
else:
|
|
assert r.status_code == 422, r.text
|
|
|
|
|
|
def test_packet_doc_filed(shop: Shop) -> None:
|
|
pkt = filed_packet(shop)
|
|
refused(add_packet_doc(shop("qc.quinn"), pkt["id"]), 403, "filed")
|
|
refused(add_packet_doc(shop("clerk.clara"), pkt["id"]), 403, "filed")
|
|
|
|
|
|
def test_packet_doc_take_out(shop: Shop) -> None:
|
|
pkt = draft(shop)
|
|
q = shop("qc.quinn")
|
|
out = ok(add_packet_doc(q, pkt["id"], "nde", "RT report W1-W4", "RPT-2210, accept"))
|
|
doc_id = int(out["docs"][-1]["id"])
|
|
for reason in (None, "", "ab"):
|
|
refused(take_out(q, pkt["id"], doc_id, reason), 400, "reason_required", MSG_TAKE_OUT)
|
|
refused(take_out(shop("wh.wade"), pkt["id"], doc_id), 403, "forbidden")
|
|
out = ok(take_out(shop("clerk.clara"), pkt["id"], doc_id, "wrong packet"))
|
|
d = doc_of(out, doc_id)
|
|
assert d["removed"]["reason"] == "wrong packet" and d["removed"]["at"]
|
|
assert d["removed"]["by"]["display_name"] == "Clara Clerk"
|
|
log = last_log(q, pkt["id"])
|
|
assert (log["action"], log["note"]) == ("Document taken out of the data book", "RT report W1-W4 · wrong packet")
|
|
refused(take_out(q, pkt["id"], doc_id, "again please"), 409, "already_removed")
|
|
row = one(shop.settings, "SELECT removed_reason, removed_by IS NOT NULL FROM packet_docs WHERE id = %s", (doc_id,))
|
|
assert tuple(row) == ("wrong packet", True)
|
|
# nothing is deleted
|
|
assert all_rows(shop.settings, "SELECT count(*) FROM packet_docs WHERE packet_id = %s", (pkt["id"],))[0][0] == 1
|
|
|
|
|
|
def test_qc_folder_docs_on_the_packet(shop: Shop) -> None:
|
|
project_id, unit_id = make_unit(shop.settings)
|
|
pkt = draft(shop, unit_id)
|
|
assert fresh(shop("qc.quinn"), pkt["id"])["qc_folder_docs"] == []
|
|
ok(make_qc_doc(shop, unit_id=unit_id, type="ITP", title="ITP unit one"))
|
|
ok(make_qc_doc(shop, project_id=project_id, type="WPS / PQR", title="WPS-101 GTAW/SMAW", rev="2"))
|
|
docs = fresh(shop("wh.wade"), pkt["id"])["qc_folder_docs"]
|
|
by_scope = {d["scope"]: d for d in docs}
|
|
assert set(by_scope) == {"unit", "project"}
|
|
assert by_scope["unit"]["type"] == "ITP" and by_scope["unit"]["title"] == "ITP unit one"
|
|
assert by_scope["project"]["rev"] == "2" and by_scope["project"]["url"].startswith("/_blob/")
|
|
|
|
|
|
# ================================================================ Q.C. folder documents
|
|
|
|
|
|
def test_qc_doc_add_unit_and_project(shop: Shop) -> None:
|
|
project_id, unit_id = make_unit(shop.settings)
|
|
pno, uno = numbers(shop, unit_id)
|
|
out = ok(make_qc_doc(shop, unit_id=unit_id, type="ITP", title="ITP unit one", file_name="itp.pdf"))
|
|
assert out["unit"]["id"] == unit_id and out["project"]["id"] == project_id
|
|
d = out["docs"][0]
|
|
assert d["type"] == "ITP" and d["unit_id"] == unit_id and d["project_id"] is None
|
|
assert d["file_name"] == "itp.pdf" and d["url"] == f"/_blob/{d['asset_id']}" and d["auto"] is False
|
|
assert d["added_by"]["display_name"] == "Quinn QC" and d["superseded_by"] is None
|
|
out = ok(
|
|
make_qc_doc(
|
|
shop, project_id=project_id, type="WPS / PQR", title="WPS-101 GTAW/SMAW", rev="2", file_name="wps.pdf"
|
|
)
|
|
)
|
|
assert out["project"]["id"] == project_id and "units" in out
|
|
assert [x["title"] for x in out["docs"]] == ["WPS-101 GTAW/SMAW"]
|
|
# the audit rows and their texts
|
|
rows = all_rows(
|
|
shop.settings, "SELECT action, subject_type, detail FROM audit_events WHERE action LIKE 'qc_doc.%%' ORDER BY id"
|
|
)
|
|
assert [r[0] for r in rows] == ["qc_doc.add", "qc_doc.add"]
|
|
assert rows[0][2]["text"] == "Q.C. document added"
|
|
assert rows[0][2]["note"] == f"{uno} · ITP · ITP unit one · itp.pdf"
|
|
assert rows[1][2]["note"] == f"{pno} · WPS / PQR · WPS-101 GTAW/SMAW Rev 2 · wps.pdf"
|
|
# the title defaults to the type
|
|
out = ok(make_qc_doc(shop, unit_id=unit_id, type="NDE procedure"))
|
|
assert out["docs"][-1]["title"] == "NDE procedure"
|
|
for t in QC_TYPES:
|
|
assert ok(make_qc_doc(shop, unit_id=unit_id, type=t))["docs"][-1]["type"] == t
|
|
|
|
|
|
def test_qc_doc_refusals(shop: Shop) -> None:
|
|
project_id, unit_id = make_unit(shop.settings)
|
|
refused(make_qc_doc(shop, unit_id=unit_id, project_id=project_id), 422, "invalid_request")
|
|
refused(make_qc_doc(shop), 422, "invalid_request")
|
|
refused(
|
|
make_qc_doc(shop, unit_id=unit_id, type="Drawing"), 422, "invalid_type", "Pick a document type from the list."
|
|
)
|
|
refused(make_qc_doc(shop, unit_id=unit_id, asset_id=csv_asset(shop("qc.quinn"))["id"]), 422, "invalid_asset")
|
|
refused(make_qc_doc(shop, unit_id=unit_id, by="wh.wade"), 403, "forbidden")
|
|
refused(make_qc_doc(shop, unit_id=unit_id, by="weld.andy"), 403, "forbidden")
|
|
refused(make_qc_doc(shop, unit_id=unit_id, by="pm.paula"), 403, "forbidden")
|
|
refused(make_qc_doc(shop, unit_id=unit_id, by="admin.ann"), 400, "it_reason_required")
|
|
assert folder_unit(shop("qc.quinn"), unit_id)["docs"] == []
|
|
# who may add: Clerical, Q.C., and an admin with an IT reason
|
|
ok(make_qc_doc(shop, unit_id=unit_id, by="clerk.clara"))
|
|
ok(make_qc_doc(shop, unit_id=unit_id, by="multi.mia"))
|
|
out = ok(make_qc_doc(shop, unit_id=unit_id, by="admin.ann", it_reason=IT))
|
|
assert len(out["docs"]) == 3
|
|
row = one(shop.settings, "SELECT it_reason FROM audit_events WHERE action = 'qc_doc.add' ORDER BY id DESC LIMIT 1")
|
|
assert row[0] == IT
|
|
# over the limits (title 140, rev 12): refused, or cut to the limit
|
|
for field, n in (("title", 140), ("rev", 12)):
|
|
r = make_qc_doc(shop, unit_id=unit_id, **{field: "x" * (n + 1)})
|
|
assert r.status_code == 422 or len(r.json()["docs"][-1][field]) <= n, (field, r.text)
|
|
|
|
|
|
def test_qc_doc_new_revision(shop: Shop) -> None:
|
|
project_id, unit_id = make_unit(shop.settings)
|
|
other_unit = add_unit(shop, project_id, "002")
|
|
uno = numbers(shop, unit_id)[1]
|
|
first = ok(make_qc_doc(shop, unit_id=unit_id, type="ITP", title="ITP L1", rev="0"))["docs"][0]
|
|
# a revision from another folder or of a missing row is refused
|
|
refused(make_qc_doc(shop, unit_id=other_unit, replaces=first["id"]), 422, "invalid_replace")
|
|
refused(make_qc_doc(shop, project_id=project_id, replaces=first["id"]), 422, "invalid_replace")
|
|
refused(make_qc_doc(shop, unit_id=unit_id, replaces=999999), 422, "invalid_replace")
|
|
out = ok(
|
|
make_qc_doc(
|
|
shop, unit_id=unit_id, type="ITP", title="ITP L1", rev="1", replaces=first["id"], file_name="itp1.pdf"
|
|
)
|
|
)
|
|
assert [d["rev"] for d in out["docs"]] == ["1"]
|
|
assert [d["id"] for d in out["superseded"]] == [first["id"]]
|
|
old = out["superseded"][0]
|
|
new_id = out["docs"][0]["id"]
|
|
assert old["superseded_by"] == new_id and old["superseded_at"]
|
|
assert old["superseded_by_user"]["display_name"] == "Quinn QC"
|
|
# a second revision of the same old row
|
|
refused(make_qc_doc(shop, unit_id=unit_id, rev="1b", replaces=first["id"]), 409, "already_superseded")
|
|
# the new one can be revised again; superseded is newest first
|
|
out = ok(make_qc_doc(shop, unit_id=unit_id, type="ITP", title="ITP L1", rev="2", replaces=new_id, by="clerk.clara"))
|
|
assert [d["rev"] for d in out["docs"]] == ["2"]
|
|
assert [d["rev"] for d in out["superseded"]] == ["1", "0"]
|
|
# nothing is ever deleted
|
|
assert all_rows(shop.settings, "SELECT count(*) FROM qc_docs WHERE unit_id = %s", (unit_id,))[0][0] == 3
|
|
rows = all_rows(shop.settings, "SELECT action, detail FROM audit_events WHERE action LIKE 'qc_doc.%%' ORDER BY id")
|
|
assert [r[0] for r in rows] == ["qc_doc.add", "qc_doc.revise", "qc_doc.revise"]
|
|
assert rows[1][1]["text"] == "Q.C. document revised"
|
|
assert rows[1][1]["note"] == f"{uno} · ITP · ITP L1 Rev 1 · itp1.pdf"
|
|
|
|
|
|
def test_qc_folders_and_counts(shop: Shop) -> None:
|
|
q = shop("qc.quinn")
|
|
project_id, unit_1 = make_unit(shop.settings)
|
|
unit_2 = add_unit(shop, project_id, "002")
|
|
closed_id, _ = make_unit(shop.settings)
|
|
run(shop.settings, "UPDATE projects SET status = 'closed' WHERE id = %s", (closed_id,))
|
|
|
|
def row(show: str, pid: int) -> dict[str, Any] | None:
|
|
rows = ok(q.get(f"/api/qc/folders?show={show}"))
|
|
return next((r for r in rows if r["id"] == pid), None)
|
|
|
|
r = row("active", project_id)
|
|
assert r is not None and r["units"] == 2 and r["docs"] == 0 and r["units_without_itp"] == 2
|
|
assert {"number", "name", "customer", "po", "status"} <= set(r)
|
|
assert row("active", closed_id) is None
|
|
closed = row("all", closed_id)
|
|
assert closed is not None and closed["units_without_itp"] == 0 and closed["status"] == "closed"
|
|
# an ITP document on unit 1, then a digital ITP on unit 2
|
|
ok(make_qc_doc(shop, unit_id=unit_1, type="ITP"))
|
|
r = row("active", project_id)
|
|
assert r is not None and r["units_without_itp"] == 1 and r["docs"] == 1
|
|
start_itp(shop, unit_2)
|
|
r = row("active", project_id)
|
|
assert r is not None and r["units_without_itp"] == 0
|
|
# a project document counts too; a WPS is not an ITP
|
|
ok(make_qc_doc(shop, project_id=project_id, type="WPS / PQR"))
|
|
r = row("active", project_id)
|
|
assert r is not None and r["docs"] == 2 and r["units_without_itp"] == 0
|
|
# the home lists the active folders
|
|
home = ok(q.get("/api/qc/home"))
|
|
assert any(f["id"] == project_id for f in home["folders"])
|
|
assert not any(f["id"] == closed_id for f in home["folders"])
|
|
numbers_in_order = [f["number"] for f in ok(q.get("/api/qc/folders?show=all"))]
|
|
assert numbers_in_order == sorted(numbers_in_order)
|
|
|
|
|
|
def test_project_itp_document_covers_every_unit(shop: Shop) -> None:
|
|
project_id, unit_1 = make_unit(shop.settings)
|
|
add_unit(shop, project_id, "002")
|
|
ok(make_qc_doc(shop, project_id=project_id, type="ITP", title="Project ITP"))
|
|
rows = ok(shop("qc.quinn").get("/api/qc/folders?show=active"))
|
|
r = next(x for x in rows if x["id"] == project_id)
|
|
assert r["units_without_itp"] == 0
|
|
assert folder_unit(shop("qc.quinn"), unit_1)["no_itp"] is False
|
|
|
|
|
|
def test_project_and_unit_folder_reads(shop: Shop) -> None:
|
|
q = shop("qc.quinn")
|
|
project_id, unit_1 = make_unit(shop.settings)
|
|
unit_2 = add_unit(shop, project_id, "002")
|
|
pno, uno = numbers(shop, unit_1)
|
|
ok(make_qc_doc(shop, project_id=project_id, type="WPS / PQR", title="WPS-101", rev="2"))
|
|
ok(make_qc_doc(shop, unit_id=unit_1, type="Customer spec", title="Spec A"))
|
|
itp = start_itp(shop, unit_2)
|
|
|
|
pf = folder_project(q, project_id)
|
|
assert pf["project"]["number"] == pno and pf["project"]["customer"] == "Acme Refining"
|
|
assert [d["title"] for d in pf["docs"]] == ["WPS-101"] and pf["superseded"] == []
|
|
units = {u["id"]: u for u in pf["units"]}
|
|
assert set(units) == {unit_1, unit_2}
|
|
assert units[unit_1]["number"] == uno and units[unit_1]["docs"] == 1 and units[unit_1]["itp"] is None
|
|
assert units[unit_1]["has_itp_doc"] is False
|
|
assert units[unit_2]["itp"]["id"] == itp["id"] and units[unit_2]["itp"]["no"] == itp["no"]
|
|
assert units[unit_2]["itp"]["total"] == 70 and units[unit_2]["itp"]["done"] == 0
|
|
|
|
uf = folder_unit(q, unit_1)
|
|
assert uf["unit"]["number"] == uno and uf["project"]["id"] == project_id
|
|
assert [d["title"] for d in uf["docs"]] == ["Spec A"]
|
|
assert [d["title"] for d in uf["inherited"]] == ["WPS-101"]
|
|
assert uf["itp"] is None and uf["no_itp"] is True
|
|
uf2 = folder_unit(q, unit_2)
|
|
assert uf2["itp"]["status"] == "open" and uf2["no_itp"] is False
|
|
# any session reads; unknown ids are 404
|
|
assert shop("wh.wade").get(f"/api/qc/folders/units/{unit_1}").status_code == 200
|
|
assert q.get("/api/qc/folders/units/999999").status_code == 404
|
|
assert q.get("/api/qc/folders/projects/999999").status_code == 404
|