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]>
677 lines
33 KiB
Python
677 lines
33 KiB
Python
"""Packet child rows: the BOM and shorts, welds, heat tags, the heat table, the stage, the warehouse board, Q.C.
|
||
and change orders. Every write goes through the pipeline; these tests check the policy, the row and state checks,
|
||
and what each write leaves behind."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import threading
|
||
from collections.abc import Iterator
|
||
from typing import Any
|
||
|
||
import psycopg
|
||
import pytest
|
||
from fakedir import FakeDirectory
|
||
from fastapi.testclient import TestClient
|
||
from pkthelpers import (
|
||
SIGN,
|
||
as_user,
|
||
finish_stage_by_sql,
|
||
get,
|
||
make_draft,
|
||
make_unit,
|
||
ok_step,
|
||
phase2_client,
|
||
ready_draft,
|
||
seed_users,
|
||
station_id,
|
||
step,
|
||
user_id,
|
||
)
|
||
from pshelpers import app_conn, needs_pg
|
||
|
||
from packetstation.settings import Settings
|
||
|
||
pytestmark = needs_pg
|
||
|
||
BOM = [
|
||
{"item": 1, "qty": 4, "part_id": "PIPE-4", "description": "4in pipe SCH40"},
|
||
{"item": 2, "qty": 2, "part_id": "FLG-4", "description": "4in flange"},
|
||
]
|
||
|
||
|
||
@pytest.fixture
|
||
def pg_client2(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[TestClient]:
|
||
with phase2_client(pg_settings, fake_dir) as client:
|
||
seed_users(client)
|
||
with app_conn(pg_settings) as conn:
|
||
conn.execute("UPDATE users SET welder_stamp = 'A7' WHERE sam_account = 'weld.andy'")
|
||
yield client
|
||
|
||
|
||
@pytest.fixture
|
||
def unit(pg_settings: Settings) -> tuple[int, int]:
|
||
with app_conn(pg_settings) as conn:
|
||
return make_unit(conn)
|
||
|
||
|
||
def _one(settings: Settings, sql: str, params: Any = ()) -> Any:
|
||
with app_conn(settings) as conn:
|
||
return conn.execute(sql, params).fetchone()
|
||
|
||
|
||
def _line(pkt: dict[str, Any], item: int) -> dict[str, Any]:
|
||
return next(b for b in pkt["bom"] if b["item"] == item)
|
||
|
||
|
||
def _im(v: int) -> dict[str, str]:
|
||
return {"If-Match": str(v)}
|
||
|
||
|
||
def warehouse_packet(client: TestClient, unit_id: int) -> dict[str, Any]:
|
||
as_user(client, "clerk.clara")
|
||
pkt = ready_draft(client, unit_id, bom=BOM)
|
||
return ok_step(client, pkt, "send_to_warehouse")
|
||
|
||
|
||
def pull_all(client: TestClient, pkt: dict[str, Any]) -> dict[str, Any]:
|
||
for b in pkt["bom"]:
|
||
if not b["pulled"] and b["short"] is None:
|
||
r = client.patch(
|
||
f"/api/packets/{pkt['id']}/bom/{b['item']}",
|
||
json={"pulled": True, "heat": f"HT-{b['item']}"},
|
||
headers=_im(b["row_version"]),
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
pkt = r.json()
|
||
return pkt
|
||
|
||
|
||
def shop_packet(client: TestClient, settings: Settings, unit_id: int) -> dict[str, Any]:
|
||
"""A packet in progress at Stall 1 with weld.andy, every BOM line pulled."""
|
||
pkt = warehouse_packet(client, unit_id)
|
||
as_user(client, "wh.wade")
|
||
pkt = pull_all(client, pkt)
|
||
pkt = ok_step(client, pkt, "wh_release")
|
||
with app_conn(settings) as conn:
|
||
stall = station_id(conn)
|
||
andy = user_id(conn, "weld.andy")
|
||
as_user(client, "sup.wes")
|
||
pkt = ok_step(client, pkt, "assign", station_id=stall, welder_id=andy)
|
||
as_user(client, "weld.andy")
|
||
pkt = ok_step(client, pkt, "start")
|
||
assert pkt["status"] == "in_progress"
|
||
return pkt
|
||
|
||
|
||
# ---------------------------------------------------------------- BOM
|
||
|
||
|
||
def test_bom_full_replace_on_a_draft(pg_client2: TestClient, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
as_user(c, "clerk.clara")
|
||
pkt = ready_draft(c, unit[1])
|
||
url = f"/api/packets/{pkt['id']}/bom"
|
||
body = {"bom": [{**BOM[0], "cl_item": "A", "material": " A106-B ", "length": "20ft"}, BOM[1]]}
|
||
assert c.put(url, json=body).status_code == 428
|
||
r = c.put(url, json=body, headers=_im(pkt["version"]))
|
||
assert r.status_code == 200, r.text
|
||
out = r.json()
|
||
assert [b["item"] for b in out["bom"]] == [1, 2] and out["version"] == pkt["version"] + 1
|
||
assert _line(out, 1)["material"] == "A106-B" and _line(out, 1)["src"] == "hand"
|
||
stale = c.put(url, json={"bom": [BOM[0]]}, headers=_im(pkt["version"]))
|
||
assert stale.status_code == 409 and stale.json()["code"] == "version_conflict", stale.text
|
||
dup = c.put(url, json={"bom": [BOM[0], BOM[0]]}, headers=_im(out["version"]))
|
||
assert dup.status_code == 422 and dup.json()["code"] == "duplicate_item"
|
||
zero = c.put(url, json={"bom": [{"item": 0, "qty": 1}]}, headers=_im(out["version"]))
|
||
assert zero.status_code == 422
|
||
r = c.put(url, json={"bom": [BOM[1]]}, headers=_im(out["version"]))
|
||
assert r.status_code == 200 and [b["item"] for b in r.json()["bom"]] == [2]
|
||
log = c.get(f"/api/packets/{pkt['id']}/log").json()
|
||
assert log[0]["action"] == "BOM replaced" and log[0]["note"] == "1 line"
|
||
|
||
wh = ok_step(c, r.json(), "send_to_warehouse")
|
||
r = c.put(url, json={"bom": BOM}, headers=_im(wh["version"]))
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden", r.text
|
||
as_user(c, "wh.wade")
|
||
r = c.put(url, json={"bom": BOM}, headers=_im(wh["version"]))
|
||
assert r.status_code == 403 and r.json()["code"] == "wrong_status", r.text
|
||
|
||
|
||
def test_bom_line_patch_in_the_warehouse(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = warehouse_packet(c, unit[1])
|
||
assert pkt["wh"]["stage"] == "incoming"
|
||
url = f"/api/packets/{pkt['id']}/bom/1"
|
||
r = c.patch(url, json={"heat": "H123"}, headers=_im(1))
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden"
|
||
assert "It's with Warehouse right now." in r.json()["detail"]
|
||
|
||
as_user(c, "wh.wade")
|
||
r = c.patch(url, json={"heat": "H123", "loc": "Rack 4"}, headers=_im(1))
|
||
assert r.status_code == 200, r.text
|
||
line = _line(r.json(), 1)
|
||
assert line["heat"] == "H123" and line["loc"] == "Rack 4" and line["row_version"] == 2
|
||
assert r.json()["wh"]["stage"] == "incoming"
|
||
stale = c.patch(url, json={"heat": "H999"}, headers=_im(1))
|
||
assert stale.status_code == 409 and stale.json()["code"] == "row_conflict"
|
||
assert stale.json()["current"]["row_version"] == 2 and stale.json()["current"]["heat"] == "H123"
|
||
bad = c.patch(url, json={"heat": "@@ bad"}, headers=_im(2))
|
||
assert bad.status_code == 422 and bad.json()["code"] == "invalid_heat"
|
||
nomtr = c.patch(url, json={"mtr_asset_id": "6f1c1d1e-0000-4000-8000-000000000000"}, headers=_im(2))
|
||
assert nomtr.status_code == 422 and nomtr.json()["code"] == "invalid_asset"
|
||
|
||
r = c.patch(url, json={"pulled": True}, headers=_im(2))
|
||
assert r.status_code == 200, r.text
|
||
out = r.json()
|
||
line = _line(out, 1)
|
||
assert line["pulled"] is True and float(line["issued"]) == 4 and line["row_version"] == 3
|
||
assert out["wh"]["stage"] == "pulling" and out["wh"]["auto"] is True and out["wh"]["note"] == "started pulling"
|
||
row = _one(
|
||
pg_settings, "SELECT pulled_at, pulled_by FROM bom_lines WHERE packet_id = %s AND item = 1", (pkt["id"],)
|
||
)
|
||
assert row[0] is not None and row[1] is not None
|
||
r = c.patch(f"/api/packets/{pkt['id']}/bom/2", json={"pulled": True}, headers=_im(1))
|
||
assert r.json()["wh"]["stage"] == "pulling"
|
||
n = _one(pg_settings, "SELECT count(*) FROM packet_wh WHERE packet_id = %s", (pkt["id"],))[0]
|
||
assert n == 2
|
||
|
||
r = c.patch(url, json={"na": True}, headers=_im(3))
|
||
assert _line(r.json(), 1)["na"] is True and _line(r.json(), 1)["heat"] is None
|
||
r = c.patch(url, json={"pulled": False}, headers=_im(4))
|
||
assert _line(r.json(), 1)["pulled"] is False
|
||
row = _one(pg_settings, "SELECT pulled_at FROM bom_lines WHERE packet_id = %s AND item = 1", (pkt["id"],))
|
||
assert row[0] is None
|
||
same = c.patch(url, json={"pulled": False}, headers=_im(5))
|
||
assert same.status_code == 200 and same.json()["version"] == r.json()["version"]
|
||
|
||
|
||
def test_wh_hold_and_first_pull_resumes(pg_client2: TestClient, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = warehouse_packet(c, unit[1])
|
||
as_user(c, "wh.wade")
|
||
r = c.post(
|
||
f"/api/packets/{pkt['id']}/wh",
|
||
json={"stage": "hold", "reason": "Waiting on purchasing / PO", "expect_stage": "incoming"},
|
||
)
|
||
assert r.status_code == 200 and r.json()["wh"]["stage"] == "hold", r.text
|
||
r = c.patch(f"/api/packets/{pkt['id']}/bom/1", json={"pulled": True}, headers=_im(1))
|
||
assert r.json()["wh"]["stage"] == "pulling" and r.json()["wh"]["note"] == "pulling resumed"
|
||
|
||
|
||
def test_shorts_and_receipts(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = warehouse_packet(c, unit[1])
|
||
pid = pkt["id"]
|
||
as_user(c, "wh.wade")
|
||
url = f"/api/packets/{pid}/bom/1/short"
|
||
body = {
|
||
"qty": 3,
|
||
"reason": "Not in stock",
|
||
"ref": "PO-991",
|
||
"eta": "2026-11-02",
|
||
"heat": "H0",
|
||
"expect_open": False,
|
||
}
|
||
r = c.post(url, json=body)
|
||
assert r.status_code == 200, r.text
|
||
out = r.json()
|
||
line = _line(out, 1)
|
||
assert float(line["short"]["qty"]) == 3 and float(line["short"]["qty_open"]) == 3
|
||
assert line["short"]["ref"] == "PO-991"
|
||
assert line["pulled"] is True and float(line["issued"]) == 1 and out["open_shorts"] == 1
|
||
assert out["wh"]["stage"] == "pulling" and out["wh"]["auto"] is True
|
||
again = c.post(url, json=body)
|
||
assert again.status_code == 409 and again.json()["code"] == "short_conflict"
|
||
assert c.post(url, json={**body, "qty": 9, "expect_open": True}).status_code == 422
|
||
r = c.post(url, json={**body, "qty": 2, "expect_open": True})
|
||
assert r.status_code == 200 and float(_line(r.json(), 1)["short"]["qty_open"]) == 2
|
||
assert float(_line(r.json(), 1)["issued"]) == 2
|
||
rv = _line(r.json(), 1)["row_version"]
|
||
blocked = c.patch(f"/api/packets/{pid}/bom/1", json={"heat": "H1"}, headers=_im(rv))
|
||
assert blocked.status_code == 409 and blocked.json()["code"] == "short_open"
|
||
|
||
# A short on item 2, then cleared: the whole quantity counts as issued.
|
||
r = c.post(f"/api/packets/{pid}/bom/2/short", json={"qty": 1, "reason": "Damaged / rejected", "expect_open": False})
|
||
assert _line(r.json(), 2)["short"] is not None
|
||
r = c.request("DELETE", f"/api/packets/{pid}/bom/2/short", json={"expect_open": True})
|
||
assert r.status_code == 200, r.text
|
||
assert _line(r.json(), 2)["short"] is None and float(_line(r.json(), 2)["issued"]) == 2
|
||
assert c.request("DELETE", f"/api/packets/{pid}/bom/2/short", json={"expect_open": True}).status_code == 409
|
||
# The release gate needs a heat on every pulled line.
|
||
rv2 = _line(r.json(), 2)["row_version"]
|
||
r = c.patch(f"/api/packets/{pid}/bom/2", json={"heat": "H2"}, headers=_im(rv2))
|
||
assert r.status_code == 200, r.text
|
||
|
||
pkt = ok_step(c, r.json(), "wh_release")
|
||
assert pkt["status"] == "ready" and pkt["open_shorts"] == 1
|
||
rec = f"/api/packets/{pid}/bom/1/short/receipts"
|
||
r = c.post(rec, json={"qty": 1, "heat": "H100", "note": "First piece"})
|
||
assert r.status_code == 200, r.text
|
||
line = _line(r.json(), 1)
|
||
assert float(line["short"]["qty_open"]) == 1 and len(line["short"]["receipts"]) == 1
|
||
assert line["short"]["receipts"][0]["by"]["display_name"] == "Wade Warehouse"
|
||
assert line["heat"] == "H0 / H100" and float(line["issued"]) == 3
|
||
assert c.post(f"/api/packets/{pid}/bom/2/short/receipts", json={"qty": 1, "heat": "H1"}).status_code == 404
|
||
|
||
with app_conn(pg_settings) as conn:
|
||
stall = station_id(conn)
|
||
andy = user_id(conn, "weld.andy")
|
||
as_user(c, "sup.wes")
|
||
pkt = ok_step(c, r.json(), "assign", station_id=stall, welder_id=andy)
|
||
as_user(c, "weld.andy")
|
||
pkt = ok_step(c, pkt, "start")
|
||
# Receive by custody: someone who neither is in the warehouse nor holds the packet is refused.
|
||
as_user(c, "qc.quinn")
|
||
no = c.post(rec, json={"qty": 1, "heat": "H200"})
|
||
assert no.status_code == 403 and no.json()["code"] == "forbidden"
|
||
as_user(c, "wh.wade")
|
||
assert c.post(rec, json={"qty": 2, "heat": "H200"}).status_code == 422
|
||
nh = c.post(rec, json={"qty": 1})
|
||
assert nh.status_code == 422 and nh.json()["code"] == "heat_required"
|
||
r = c.post(rec, json={"qty": 1, "heat": "H200"})
|
||
assert r.status_code == 200, r.text
|
||
line = _line(r.json(), 1)
|
||
assert line["short"] is None and r.json()["open_shorts"] == 0 and r.json()["status"] == "in_progress"
|
||
assert line["heat"] == "H0 / H100 / H200" and float(line["issued"]) == 4
|
||
closed = _one(
|
||
pg_settings,
|
||
"SELECT s.closed_at, s.qty_open FROM bom_shorts s JOIN bom_lines b ON b.id = s.bom_line_id"
|
||
" WHERE s.packet_id = %s AND b.item = 1",
|
||
(pid,),
|
||
)
|
||
assert closed[0] is not None and closed[1] == 0
|
||
with app_conn(pg_settings) as conn, pytest.raises(psycopg.Error):
|
||
conn.execute("UPDATE short_receipts SET qty = 9 WHERE packet_id = %s", (pid,))
|
||
with app_conn(pg_settings) as conn, pytest.raises(psycopg.Error):
|
||
conn.execute("DELETE FROM short_receipts WHERE packet_id = %s", (pid,))
|
||
|
||
|
||
# ---------------------------------------------------------------- welds, heat tags, heat table
|
||
|
||
|
||
def test_welds(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = shop_packet(c, pg_settings, unit[1])
|
||
pid = pkt["id"]
|
||
weld = {"page": 1, "fx": 0.25, "fy": 0.5, "wo_base": "L670001", "piece": 1, "items": [2, 1], "type": "Fillet"}
|
||
r1 = c.post(f"/api/packets/{pid}/welds", json={**weld, "size": "1/4"})
|
||
assert r1.status_code == 200, r1.text
|
||
r2 = c.post(f"/api/packets/{pid}/welds", json={**weld, "type": "Groove"})
|
||
assert r2.status_code == 200, r2.text
|
||
welds = r2.json()["welds"]
|
||
assert [w["weld_no"] for w in welds] == [1, 2]
|
||
assert welds[0]["items"] == [1, 2] and welds[0]["stamp"] == "A7"
|
||
assert welds[0]["welder"]["display_name"] == "Andy Welder"
|
||
row = _one(pg_settings, "SELECT dept_code, stage_id FROM welds WHERE packet_id = %s AND weld_no = 1", (pid,))
|
||
assert row[0] == "03" and row[1] == pkt["stage"]["id"]
|
||
bad = c.post(f"/api/packets/{pid}/welds", json={**weld, "items": [9]})
|
||
assert bad.status_code == 422 and bad.json()["code"] == "invalid_items"
|
||
assert c.post(f"/api/packets/{pid}/welds", json={**weld, "items": []}).status_code == 422
|
||
assert c.post(f"/api/packets/{pid}/welds", json={**weld, "stamp": "no stamp!"}).status_code == 422
|
||
|
||
r = c.patch(f"/api/packets/{pid}/welds/1", json={"size": "3/8"}, headers=_im(1))
|
||
assert r.status_code == 200 and r.json()["welds"][0]["size"] == "3/8" and r.json()["welds"][0]["row_version"] == 2
|
||
stale = c.patch(f"/api/packets/{pid}/welds/1", json={"size": "1/2"}, headers=_im(1))
|
||
assert stale.status_code == 409 and stale.json()["code"] == "row_conflict"
|
||
r = c.request("DELETE", f"/api/packets/{pid}/welds/2", json={"reason": "Marked twice"})
|
||
assert r.status_code == 200 and [w["weld_no"] for w in r.json()["welds"]] == [1]
|
||
row = _one(pg_settings, "SELECT removed_reason FROM welds WHERE packet_id = %s AND weld_no = 2", (pid,))
|
||
assert row[0] == "Marked twice"
|
||
assert c.patch(f"/api/packets/{pid}/welds/2", json={"size": "1"}, headers=_im(1)).status_code == 404
|
||
r = c.post(f"/api/packets/{pid}/welds", json=weld)
|
||
assert r.json()["welds"][-1]["weld_no"] == 3
|
||
|
||
# Another worker of the department, limited to another stall, does not hold this packet.
|
||
with app_conn(pg_settings) as conn:
|
||
conn.execute(
|
||
"INSERT INTO user_station_limits (user_id, station_id) VALUES (%s, %s)",
|
||
(user_id(conn, "dup.dave"), station_id(conn, "03", "Stall 2")),
|
||
)
|
||
as_user(c, "dup.dave")
|
||
r = c.post(f"/api/packets/{pid}/welds", json=weld)
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden", r.text
|
||
as_user(c, "clerk.clara")
|
||
assert c.post(f"/api/packets/{pid}/welds", json=weld).status_code == 403
|
||
|
||
|
||
def test_heat_tags_and_heat_table(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = warehouse_packet(c, unit[1])
|
||
pid = pkt["id"]
|
||
assert c.post(f"/api/packets/{pid}/heat-tags", json={"page": 1, "item": 1}).status_code == 403
|
||
as_user(c, "wh.wade")
|
||
for item in (1, 2):
|
||
r = c.post(f"/api/packets/{pid}/heat-tags", json={"page": 1, "item": item, "fx": 0.1, "fy": 0.2})
|
||
assert r.status_code == 200, r.text
|
||
tags = r.json()["heat_tags"]
|
||
assert [(t["tag_no"], t["item"]) for t in tags] == [(1, 1), (2, 2)]
|
||
assert c.post(f"/api/packets/{pid}/heat-tags", json={"page": 1, "item": 7}).status_code == 422
|
||
r = c.patch(f"/api/packets/{pid}/heat-tags/2", json={"item": 1}, headers=_im(1))
|
||
assert r.status_code == 200 and r.json()["heat_tags"][1]["item"] == 1
|
||
stale = c.patch(f"/api/packets/{pid}/heat-tags/2", json={"item": 2}, headers=_im(1))
|
||
assert stale.status_code == 409 and stale.json()["code"] == "row_conflict"
|
||
r = c.request("DELETE", f"/api/packets/{pid}/heat-tags/1", json={})
|
||
assert r.status_code == 200 and [t["tag_no"] for t in r.json()["heat_tags"]] == [2]
|
||
|
||
url = f"/api/packets/{pid}/heat-table/1"
|
||
r1 = c.put(url, json={"x": 0.7, "y": 0.1, "s": 1.0})
|
||
assert r1.status_code == 200
|
||
ht = r1.json()["heat_table"]
|
||
assert [{k: h[k] for k in ("page", "x", "y", "s")} for h in ht] == [{"page": 1, "x": 0.7, "y": 0.1, "s": 1.0}]
|
||
assert ht[0]["carried_from"] is None and ht[0]["by"]["display_name"] == "Wade Warehouse"
|
||
r2 = c.put(url, json={"x": 0.7, "y": 0.1, "s": 1.0})
|
||
assert r2.status_code == 200 and r2.json()["version"] == r1.json()["version"]
|
||
r3 = c.put(url, json={"x": 0.6, "y": 0.1, "s": 1.2})
|
||
assert r3.json()["heat_table"][0]["x"] == 0.6 and r3.json()["version"] == r1.json()["version"] + 1
|
||
n = _one(pg_settings, "SELECT count(*) FROM heat_table_placements WHERE packet_id = %s", (pid,))[0]
|
||
assert n == 1
|
||
assert c.put(url, json={"x": 0.6, "y": 0.1, "s": 0}).status_code == 422
|
||
r = c.request("DELETE", url, json={})
|
||
assert r.status_code == 200 and r.json()["heat_table"] == []
|
||
assert c.request("DELETE", url, json={}).status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------- the stage
|
||
|
||
|
||
def test_stage_fields_and_checklist(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = shop_packet(c, pg_settings, unit[1])
|
||
pid = pkt["id"]
|
||
rv = pkt["stage"]["row_version"]
|
||
r = c.patch(f"/api/packets/{pid}/stage", json={"filler": "ER70S-6", "preheat": "50F"}, headers=_im(rv))
|
||
assert r.status_code == 200, r.text
|
||
st = r.json()["stage"]
|
||
assert st["filler"] == "ER70S-6" and st["preheat"] == "50F" and st["row_version"] == rv + 1
|
||
stale = c.patch(f"/api/packets/{pid}/stage", json={"filler": "E7018"}, headers=_im(rv))
|
||
assert stale.status_code == 409 and stale.json()["code"] == "row_conflict"
|
||
|
||
andy = user_id_of(pg_settings, "weld.andy")
|
||
r = c.put(f"/api/packets/{pid}/stage/checklist/dwg", json={"done": True})
|
||
assert r.status_code == 200, r.text
|
||
entry = r.json()["stage"]["checklist"]["dwg"]
|
||
assert entry["done"] is True and entry["by"] == andy and entry["stamp"] == "A7" and entry["at"]
|
||
assert r.json()["stage"]["row_version"] == st["row_version"]
|
||
bad = c.put(f"/api/packets/{pid}/stage/checklist/nope", json={"done": True})
|
||
assert bad.status_code == 422 and bad.json()["code"] == "invalid_key"
|
||
as_user(c, "lead.lou")
|
||
r = c.put(f"/api/packets/{pid}/stage/checklist/heat", json={"done": True})
|
||
assert r.status_code == 200 and r.json()["stage"]["checklist"]["heat"]["by"] == user_id_of(pg_settings, "lead.lou")
|
||
r = c.put(f"/api/packets/{pid}/stage/checklist/dwg", json={"done": False})
|
||
assert r.json()["stage"]["checklist"]["dwg"] == {"done": False}
|
||
as_user(c, "wh.wade")
|
||
assert c.put(f"/api/packets/{pid}/stage/checklist/dwg", json={"done": True}).status_code == 403
|
||
|
||
|
||
def user_id_of(settings: Settings, sam: str) -> str:
|
||
with app_conn(settings) as conn:
|
||
return user_id(conn, sam)
|
||
|
||
|
||
def test_stage_clock(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = shop_packet(c, pg_settings, unit[1])
|
||
url = f"/api/packets/{pkt['id']}/stage/clock"
|
||
assert pkt["stage"]["clock"]["running_since"] is not None
|
||
r = c.post(url, json={"op": "start"})
|
||
assert r.status_code == 409 and r.json()["code"] == "clock_state"
|
||
r = c.post(url, json={"op": "pause"})
|
||
assert r.status_code == 200 and r.json()["stage"]["clock"]["running_since"] is None, r.text
|
||
assert r.json()["stage"]["clock"]["elapsed_ms"] >= 0
|
||
assert c.post(url, json={"op": "pause"}).json()["code"] == "clock_state"
|
||
r = c.post(url, json={"op": "resume"})
|
||
assert r.status_code == 200 and r.json()["stage"]["clock"]["running_since"] is not None
|
||
assert c.post(url, json={"op": "resume"}).status_code == 409
|
||
assert c.post(url, json={"op": "stop"}).status_code == 422
|
||
as_user(c, "clerk.clara")
|
||
assert c.post(url, json={"op": "pause"}).status_code == 403
|
||
|
||
|
||
def test_no_stage(pg_client2: TestClient, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
as_user(c, "clerk.clara")
|
||
pkt = make_draft(c, unit[1])
|
||
r = c.post(f"/api/packets/{pkt['id']}/stage/clock", json={"op": "start"})
|
||
assert r.status_code == 403 and r.json()["code"] == "wrong_status"
|
||
|
||
|
||
# ---------------------------------------------------------------- the warehouse board
|
||
|
||
|
||
def test_wh_stage_board(pg_client2: TestClient, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = warehouse_packet(c, unit[1])
|
||
url = f"/api/packets/{pkt['id']}/wh"
|
||
r = c.post(url, json={"stage": "pulling", "expect_stage": "incoming"})
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden"
|
||
as_user(c, "wh.wade")
|
||
r = c.post(url, json={"stage": "pulling", "expect_stage": "incoming"})
|
||
assert r.status_code == 200 and r.json()["wh"]["stage"] == "pulling" and r.json()["wh"]["auto"] is False
|
||
assert r.json()["wh"]["by"]["display_name"] == "Wade Warehouse"
|
||
stale = c.post(url, json={"stage": "pulling", "expect_stage": "incoming"})
|
||
assert stale.status_code == 409 and stale.json()["code"] == "stale_step"
|
||
assert stale.json()["current"]["stage"] == "pulling"
|
||
r = c.post(url, json={"stage": "hold", "expect_stage": "pulling"})
|
||
assert r.status_code == 400 and r.json()["code"] == "invalid_reason"
|
||
r = c.post(url, json={"stage": "hold", "reason": "Other", "expect_stage": "pulling"})
|
||
assert r.status_code == 400 and r.json()["code"] == "reason_required"
|
||
hold = {"stage": "hold", "reason": "Drawing / BOM problem", "note": "Asked Leo", "expect_stage": "pulling"}
|
||
r = c.post(url, json=hold)
|
||
assert r.status_code == 200 and r.json()["wh"]["reason"] == "Drawing / BOM problem"
|
||
assert r.json()["wh"]["note"] == "Asked Leo"
|
||
bad = c.post(url, json={"stage": "pulled", "expect_stage": "hold"})
|
||
assert bad.status_code == 409 and bad.json()["code"] == "wh_transition"
|
||
r = c.post(url, json={"stage": "pulling", "expect_stage": "hold"})
|
||
assert r.status_code == 200
|
||
r = c.post(url, json={"stage": "pulled", "expect_stage": "pulling"})
|
||
assert r.status_code == 200 and r.json()["wh"]["stage"] == "pulled"
|
||
log = c.get(f"/api/packets/{pkt['id']}/log").json()
|
||
assert log[0]["action"] == "Warehouse: Marked pulled, ready for production"
|
||
assert log[1]["action"] == "Warehouse: Pulling resumed"
|
||
|
||
|
||
# ---------------------------------------------------------------- Q.C.
|
||
|
||
|
||
def test_qc_patch(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = shop_packet(c, pg_settings, unit[1])
|
||
pid = pkt["id"]
|
||
r = c.patch(f"/api/packets/{pid}/qc", json={"inspector": "Quinn"}, headers=_im(1))
|
||
assert r.status_code == 403
|
||
finish_stage_by_sql(pg_settings, pid)
|
||
pkt = ok_step(c, get(c, pid), "submit", sign=SIGN)
|
||
as_user(c, "sup.wes")
|
||
pkt = ok_step(c, pkt, "send_to_qc", book=True, spool=True, location="Bay 3", sign=SIGN)
|
||
rv = pkt["qc"]["row_version"]
|
||
as_user(c, "qc.quinn")
|
||
url = f"/api/packets/{pid}/qc"
|
||
body = {"checks": {"db_heats": True}, "weld_exam": {"1": {"result": "accept"}}, "inspector": "Quinn QC"}
|
||
r = c.patch(url, json={**body, "stamp": "Q5"}, headers=_im(rv))
|
||
assert r.status_code == 200, r.text
|
||
qc = r.json()["qc"]
|
||
assert qc["checks"]["db_heats"]["done"] is True and qc["checks"]["db_heats"]["stamp"] == "Q5"
|
||
assert qc["checks"]["db_heats"]["by"] == user_id_of(pg_settings, "qc.quinn")
|
||
assert qc["weld_exam"]["1"]["result"] == "accept" and qc["inspector"] == "Quinn QC"
|
||
assert qc["row_version"] == rv + 1
|
||
r = c.patch(
|
||
url, json={"checks": {"db_mtr": True}, "weld_exam": {"1": {"note": "Good tie-in"}}}, headers=_im(rv + 1)
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
qc = r.json()["qc"]
|
||
assert qc["checks"]["db_heats"]["done"] and qc["checks"]["db_mtr"]["done"]
|
||
assert qc["weld_exam"]["1"]["result"] == "accept"
|
||
assert qc["weld_exam"]["1"]["note"] == "Good tie-in"
|
||
stale = c.patch(url, json={"notes": "x"}, headers=_im(rv))
|
||
assert stale.status_code == 409 and stale.json()["code"] == "row_conflict"
|
||
assert c.patch(url, json={"checks": {"nope": True}}, headers=_im(rv + 2)).status_code == 422
|
||
assert c.patch(url, json={"weld_exam": {"1": {"result": "maybe"}}}, headers=_im(rv + 2)).status_code == 422
|
||
assert c.patch(url, json={"weld_exam": {"8": {"result": "accept"}}}, headers=_im(rv + 2)).status_code == 422
|
||
r = c.patch(url, json={"checks": {"db_mtr": False}}, headers=_im(rv + 2))
|
||
assert r.json()["qc"]["checks"]["db_mtr"] == {"done": False}
|
||
as_user(c, "wh.wade")
|
||
assert c.patch(url, json={"notes": "x"}, headers=_im(rv + 3)).status_code == 403
|
||
|
||
|
||
# ---------------------------------------------------------------- change orders
|
||
|
||
|
||
def test_change_orders(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = shop_packet(c, pg_settings, unit[1])
|
||
pid = pkt["id"]
|
||
finish_stage_by_sql(pg_settings, pid)
|
||
as_user(c, "clerk.clara")
|
||
body = {"reason": "Rev C moves nozzle N2", "ecn": "ECN-55", "items": [{"old_page": 1, "new_page": 1}]}
|
||
r = c.post(f"/api/packets/{pid}/change-orders", json=body)
|
||
assert r.status_code == 400 and r.json()["code"] == "signature_required"
|
||
r = c.post(f"/api/packets/{pid}/change-orders", json={**body, "sign": SIGN})
|
||
assert r.status_code == 200, r.text
|
||
co = r.json()["change_orders"][0]
|
||
assert co["co_no"] == 1 and co["origin"] == "manual" and co["review"] == "approved"
|
||
assert co["required"] == ["03", "WH"] and co["issued_by"]["display_name"] == "Clara Clerk"
|
||
assert co["items"][0]["old_page"] == 1 and r.json()["open_cos"] == [1]
|
||
bad = c.post(f"/api/packets/{pid}/change-orders", json={**body, "required_depts": ["CL"], "sign": SIGN})
|
||
assert bad.status_code == 422 and bad.json()["code"] == "invalid_dept"
|
||
listed = c.get(f"/api/packets/{pid}/change-orders")
|
||
assert listed.status_code == 200 and [x["co_no"] for x in listed.json()] == [1]
|
||
|
||
as_user(c, "weld.andy")
|
||
r = step(c, get(c, pid), "submit", sign=SIGN)
|
||
assert r.status_code == 403 and r.json()["code"] == "co_hold", r.text
|
||
assert "Change order CO-1" in r.json()["detail"]
|
||
vurl = f"/api/packets/{pid}/change-orders/1/verify"
|
||
assert c.post(vurl, json={"dept": "03", "sign": SIGN}).status_code == 403
|
||
|
||
as_user(c, "sup.wes")
|
||
r = c.post(vurl, json={"dept": "03", "note": "Checked N2", "sign": SIGN})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["change_orders"][0]["verified"]["03"]["note"] == "Checked N2"
|
||
assert r.json()["open_cos"] == [1]
|
||
again = c.post(vurl, json={"dept": "03", "sign": SIGN})
|
||
assert again.status_code == 409 and again.json()["code"] == "already_verified"
|
||
assert c.post(vurl, json={"dept": "WH", "sign": SIGN}).status_code == 403
|
||
|
||
as_user(c, "wh.wade")
|
||
r = c.post(vurl, json={"dept": "WH", "sign": SIGN})
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden"
|
||
|
||
as_user(c, "admin.ann")
|
||
r = c.post(vurl, json={"dept": "WH", "sign": SIGN})
|
||
assert r.status_code == 400 and r.json()["code"] == "it_reason_required"
|
||
r = c.post(vurl, json={"dept": "QC", "sign": SIGN, "it_reason": "Warehouse lead is out today"})
|
||
assert r.status_code == 422 and r.json()["code"] == "invalid_dept"
|
||
r = c.post(vurl, json={"dept": "WH", "sign": SIGN, "it_reason": "Warehouse lead is out today"})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["open_cos"] == []
|
||
row = _one(pg_settings, "SELECT it_reason FROM co_verifications WHERE dept_code = 'WH'")
|
||
assert row[0] == "Warehouse lead is out today"
|
||
|
||
as_user(c, "weld.andy")
|
||
r = step(c, get(c, pid), "submit", sign=SIGN)
|
||
assert r.status_code == 200 and r.json()["status"] == "review", r.text
|
||
|
||
# A pending change from VISUAL holds nothing until Clerical holds it.
|
||
with app_conn(pg_settings) as conn:
|
||
conn.execute(
|
||
"INSERT INTO change_orders (packet_id, co_no, origin, review, reason) VALUES (%s, 2, 'visual', 'pending',"
|
||
" 'Drawing rev changed in VISUAL')",
|
||
(pid,),
|
||
)
|
||
conn.execute("UPDATE packets SET next_co_no = 3 WHERE id = %s", (pid,))
|
||
as_user(c, "clerk.clara")
|
||
assert get(c, pid)["open_cos"] == []
|
||
hurl = f"/api/packets/{pid}/change-orders/2/hold"
|
||
r = c.post(hurl, json={"hold": True})
|
||
assert r.status_code == 200 and r.json()["open_cos"] == [2], r.text
|
||
assert r.json()["change_orders"][1]["hold"] is True
|
||
as_user(c, "sup.wes")
|
||
r = step(c, get(c, pid), "send_to_qc", book=True, spool=True, location="Bay 3", sign=SIGN)
|
||
assert r.status_code == 403 and r.json()["code"] == "co_hold"
|
||
as_user(c, "clerk.clara")
|
||
r = c.post(hurl, json={"hold": False})
|
||
assert r.status_code == 200 and r.json()["open_cos"] == []
|
||
assert c.post(f"/api/packets/{pid}/change-orders/1/hold", json={"hold": True}).json()["code"] == "co_state"
|
||
rurl = f"/api/packets/{pid}/change-orders/2/reject"
|
||
r = c.post(rurl, json={"sign": SIGN})
|
||
assert r.status_code == 400 and r.json()["code"] == "reason_required"
|
||
r = c.post(rurl, json={"reason": "Not a real change", "sign": SIGN})
|
||
assert r.status_code == 200, r.text
|
||
co2 = r.json()["change_orders"][1]
|
||
assert co2["review"] == "rejected" and co2["rejected_reason"] == "Not a real change"
|
||
r = c.post(f"/api/packets/{pid}/change-orders/2/approve", json={"sign": SIGN})
|
||
assert r.status_code == 409 and r.json()["code"] == "co_state"
|
||
assert c.post(f"/api/packets/{pid}/change-orders/9/approve", json={"sign": SIGN}).status_code == 404
|
||
|
||
|
||
def test_co_approve_pending(pg_client2: TestClient, pg_settings: Settings, unit: tuple[int, int]) -> None:
|
||
c = pg_client2
|
||
pkt = warehouse_packet(c, unit[1])
|
||
pid = pkt["id"]
|
||
with app_conn(pg_settings) as conn:
|
||
conn.execute(
|
||
"INSERT INTO change_orders (packet_id, co_no, origin, review) VALUES (%s, 1, 'visual', 'pending')", (pid,)
|
||
)
|
||
conn.execute("UPDATE packets SET next_co_no = 2 WHERE id = %s", (pid,))
|
||
r = c.post(f"/api/packets/{pid}/change-orders/1/approve", json={"sign": SIGN})
|
||
assert r.status_code == 200, r.text
|
||
co = r.json()["change_orders"][0]
|
||
assert co["review"] == "approved" and co["approved_by"]["display_name"] == "Clara Clerk"
|
||
assert co["required"] == ["WH"] and r.json()["open_cos"] == [1]
|
||
|
||
|
||
# ---------------------------------------------------------------- concurrency
|
||
|
||
|
||
def test_concurrent_welds_get_distinct_numbers(
|
||
pg_client2: TestClient, pg_settings: Settings, fake_dir: FakeDirectory, unit: tuple[int, int]
|
||
) -> None:
|
||
pkt = shop_packet(pg_client2, pg_settings, unit[1])
|
||
with phase2_client(pg_settings, fake_dir) as other:
|
||
as_user(other, "weld.andy")
|
||
results: list[Any] = [None, None]
|
||
barrier = threading.Barrier(2)
|
||
|
||
def run(i: int, client: TestClient) -> None:
|
||
barrier.wait()
|
||
results[i] = client.post(f"/api/packets/{pkt['id']}/welds", json={"items": [1], "type": "Fillet"})
|
||
|
||
threads = [threading.Thread(target=run, args=(i, cl)) for i, cl in enumerate((pg_client2, other))]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join(30)
|
||
assert [r.status_code for r in results] == [200, 200], [r.text for r in results]
|
||
nos = [r[0] for r in _all(pg_settings, "SELECT weld_no FROM welds WHERE packet_id = %s", (pkt["id"],))]
|
||
assert sorted(nos) == [1, 2]
|
||
|
||
|
||
def test_concurrent_bom_patch_one_wins(
|
||
pg_client2: TestClient, pg_settings: Settings, fake_dir: FakeDirectory, unit: tuple[int, int]
|
||
) -> None:
|
||
pkt = warehouse_packet(pg_client2, unit[1])
|
||
as_user(pg_client2, "wh.wade")
|
||
with phase2_client(pg_settings, fake_dir) as other:
|
||
as_user(other, "wh.wade")
|
||
results: list[Any] = [None, None]
|
||
barrier = threading.Barrier(2)
|
||
|
||
def run(i: int, client: TestClient) -> None:
|
||
barrier.wait()
|
||
results[i] = client.patch(f"/api/packets/{pkt['id']}/bom/1", json={"heat": f"H{i}"}, headers=_im(1))
|
||
|
||
threads = [threading.Thread(target=run, args=(i, cl)) for i, cl in enumerate((pg_client2, other))]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join(30)
|
||
codes = sorted(r.status_code for r in results)
|
||
assert codes == [200, 409], [r.text for r in results]
|
||
loser = next(r for r in results if r.status_code == 409)
|
||
assert loser.json()["code"] == "row_conflict" and loser.json()["current"]["row_version"] == 2
|
||
|
||
|
||
def _all(settings: Settings, sql: str, params: Any = ()) -> list[Any]:
|
||
with app_conn(settings) as conn:
|
||
return conn.execute(sql, params).fetchall()
|