spencermandClaude Fable 5.1
39a4eb5213
The Warehouse home with the numbers, the Open shorts table, the find bar and the five-column task board (waiting, being pulled, on hold, pulled, released in the last 3 days) with buttons, hold reasons and drag. Every role sees the warehouse stage on a packet as the WH pill and the line under it. The pull table on the Materials tab: Pulled with auto-start, issued, location, heat with Enter to the next line, shorts with receipts, N/A, MTR per line with a viewer. Send back to Clerical with a reason. Release to supervisor stays blocked until every line is pulled or short and every pulled line has a heat. Server: the heats_recorded precondition on WH_RELEASE, receive by custody, server-validated hold and short reasons, the old packet history texts, GET /api/warehouse/board, cards with the warehouse stage and pull counts. Uploads typed by content per kind (drawing, mtr, document, bom with XLSX and CSV), limits per kind, optional ClamAV behind the clamav compose profile, a storage quota alert on the new notification_events table (migration 0005) shown on the Admin Storage tab. Tests: every warehouse action as every role, the release gate, the short-receive exception, the heat pattern, upload sniffing and limits, the ClamAV path with a fake scanner, the quota alert; frontend board and pull table tests; Playwright scenarios for the board and the pull flow. Verified on docker-1 in the CI stack: 3557 server tests, 106 frontend tests, 17 Playwright scenarios. Docs updated; D5 closed. Co-Authored-By: Claude Fable 5.1 <[email protected]>
781 lines
37 KiB
Python
781 lines
37 KiB
Python
"""Phase 4 warehouse API: every warehouse action as every role, hold reasons, the Pulled auto-start, the heat
|
||
pattern, the release gate (pulled-or-short and heats), receiving a short by custody, the board endpoint, the card
|
||
fields, and the exact packet_log texts and audit rows of each write."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import struct
|
||
import zlib
|
||
from collections.abc import Callable, Iterator
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from fakedir import FakeDirectory
|
||
from fastapi.testclient import TestClient
|
||
from pkthelpers import as_user, ok_step, phase2_client, ready_draft, seed_users, station_id, step, user_id
|
||
from pshelpers import ORIGIN_HEADERS, app_conn, needs_pg, sign_in
|
||
|
||
from packetstation.settings import Settings
|
||
|
||
pytestmark = needs_pg
|
||
|
||
IT = "Ticket 5120: warehouse lead is out"
|
||
MSG_PULL = "Pull every material line or mark it short first."
|
||
MSG_HEAT = "Scan or type the heat number of every pulled line first. Use N/A for a part with no traceability."
|
||
HOLD_REASONS = (
|
||
"Waiting on material (not in stock)",
|
||
"Waiting on purchasing / PO",
|
||
"Question for engineering or clerical",
|
||
"Drawing / BOM problem",
|
||
"Other",
|
||
)
|
||
# Everyone but the warehouse and the admin, on a packet the warehouse holds.
|
||
REFUSED_IN_WH = ("clerk.clara", "qc.quinn", "pm.paula", "weld.andy", "lead.lou", "sup.wes")
|
||
# On a ready packet, lead.lou and sup.wes hold it and may receive; these may not.
|
||
REFUSED_IN_READY = ("clerk.clara", "qc.quinn", "pm.paula", "weld.andy")
|
||
|
||
# A normal line, an N/A line (set by SQL), a sub-assembly line (sub = 'Y' by SQL) and a line with qty 4.
|
||
BOM = [
|
||
{"item": 1, "qty": 1, "part_id": "PIPE-4", "description": "4in pipe SCH40", "material": "A106-B", "length": "20ft"},
|
||
{"item": 2, "qty": 2, "part_id": "GSKT-4", "description": "4in gasket"},
|
||
{"item": 3, "qty": 1, "part_id": "SUB-7", "description": "Nozzle sub-assembly"},
|
||
{"item": 4, "qty": 4, "part_id": "STUD-58", "description": "5/8 stud bolt"},
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------- helpers
|
||
|
||
|
||
@dataclass
|
||
class Shop:
|
||
"""One app, one signed-in TestClient per user (separate cookie jars)."""
|
||
|
||
main: TestClient
|
||
settings: Settings
|
||
clients: dict[str, TestClient] = field(default_factory=dict)
|
||
|
||
def __call__(self, sam: str) -> TestClient:
|
||
c = self.clients.get(sam)
|
||
if c is None:
|
||
c = TestClient(self.main.app, headers=ORIGIN_HEADERS)
|
||
r = sign_in(c, sam)
|
||
assert r.status_code == 200, r.text
|
||
self.clients[sam] = c
|
||
return c
|
||
|
||
|
||
@pytest.fixture
|
||
def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
|
||
with phase2_client(pg_settings, fake_dir) as client:
|
||
seed_users(client)
|
||
yield Shop(client, pg_settings)
|
||
|
||
|
||
def _one(settings: Settings, sql: str, params: Any = ()) -> Any:
|
||
with app_conn(settings) as conn:
|
||
return conn.execute(sql, params).fetchone()
|
||
|
||
|
||
def _all(settings: Settings, sql: str, params: Any = ()) -> list[Any]:
|
||
with app_conn(settings) as conn:
|
||
return conn.execute(sql, params).fetchall()
|
||
|
||
|
||
def _im(v: int) -> dict[str, str]:
|
||
return {"If-Match": str(v)}
|
||
|
||
|
||
def line(pkt: dict[str, Any], item: int) -> dict[str, Any]:
|
||
return next(b for b in pkt["bom"] if b["item"] == item)
|
||
|
||
|
||
def fresh(c: TestClient, pid: int) -> dict[str, Any]:
|
||
r = c.get(f"/api/packets/{pid}")
|
||
assert r.status_code == 200, r.text
|
||
out: dict[str, Any] = r.json()
|
||
return out
|
||
|
||
|
||
def log_rows(c: TestClient, pid: int) -> list[dict[str, Any]]:
|
||
r = c.get(f"/api/packets/{pid}/log")
|
||
assert r.status_code == 200, r.text
|
||
out: list[dict[str, Any]] = r.json()
|
||
return out
|
||
|
||
|
||
def last_log(c: TestClient, pid: int) -> dict[str, Any]:
|
||
return log_rows(c, pid)[0]
|
||
|
||
|
||
def last_audit(settings: Settings, pid: int) -> Any:
|
||
return _one(
|
||
settings,
|
||
"SELECT action, detail, it_reason FROM audit_events WHERE packet_id = %s ORDER BY id DESC LIMIT 1",
|
||
(pid,),
|
||
)
|
||
|
||
|
||
def audit_count(settings: Settings, pid: int) -> int:
|
||
return int(_one(settings, "SELECT count(*) FROM audit_events WHERE packet_id = %s", (pid,))[0])
|
||
|
||
|
||
def make_unit_sql(settings: Settings, number: str, lot: str = "001") -> tuple[int, int]:
|
||
with app_conn(settings) as conn:
|
||
pid = conn.execute(
|
||
"INSERT INTO projects (number, name, customer, po) VALUES (%s, 'Test project', 'Acme Refining', 'PO-77')"
|
||
" RETURNING id",
|
||
(number,),
|
||
).fetchone()[0]
|
||
uid = conn.execute(
|
||
"INSERT INTO units (project_id, lot, name, status) VALUES (%s, %s, 'Unit one', 'open') RETURNING id",
|
||
(pid, lot),
|
||
).fetchone()[0]
|
||
return int(pid), int(uid)
|
||
|
||
|
||
def warehouse_packet(shop: Shop, unit_id: int, **over: Any) -> dict[str, Any]:
|
||
"""A hand draft by clerk.clara with the four BOM lines, drawing attached, heat map answered, sent to the
|
||
warehouse. Item 2 is N/A and item 3 a sub-assembly (both by SQL)."""
|
||
c = shop("clerk.clara")
|
||
pkt = ready_draft(c, unit_id, bom=BOM, **over)
|
||
with app_conn(shop.settings) as conn:
|
||
conn.execute("UPDATE bom_lines SET na = true WHERE packet_id = %s AND item = 2", (pkt["id"],))
|
||
conn.execute("UPDATE bom_lines SET sub = 'Y' WHERE packet_id = %s AND item = 3", (pkt["id"],))
|
||
pkt = fresh(c, pkt["id"])
|
||
out = ok_step(c, pkt, "send_to_warehouse")
|
||
assert out["status"] == "warehouse" and out["wh"]["stage"] == "incoming"
|
||
return out
|
||
|
||
|
||
def tiny_png() -> bytes:
|
||
def chunk(kind: bytes, data: bytes) -> bytes:
|
||
return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
|
||
|
||
ihdr = struct.pack(">IIBBBBB", 1, 1, 8, 2, 0, 0, 0)
|
||
idat = zlib.compress(b"\x00\xff\x00\x00")
|
||
return b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr) + chunk(b"IDAT", idat) + chunk(b"IEND", b"")
|
||
|
||
|
||
def upload_mtr(c: TestClient, data: bytes, name: str = "mtr.png", mime: str = "image/png") -> dict[str, Any]:
|
||
r = c.post("/api/assets", files={"file": (name, data, mime)}, data={"purpose": "mtr"})
|
||
assert r.status_code == 201, r.text
|
||
out: dict[str, Any] = r.json()
|
||
return out
|
||
|
||
|
||
def _it(body: dict[str, Any], it: str | None) -> dict[str, Any]:
|
||
return {**body, "it_reason": it} if it else body
|
||
|
||
|
||
Call = Callable[[TestClient, dict[str, Any], str | None], Any]
|
||
|
||
|
||
def wh_move(stage: str, expect: str, **kw: Any) -> Call:
|
||
def call(c: TestClient, pkt: dict[str, Any], it: str | None = None) -> Any:
|
||
return c.post(f"/api/packets/{pkt['id']}/wh", json=_it({"stage": stage, "expect_stage": expect, **kw}, it))
|
||
|
||
return call
|
||
|
||
|
||
def bom_patch(item: int, **fields: Any) -> Call:
|
||
def call(c: TestClient, pkt: dict[str, Any], it: str | None = None) -> Any:
|
||
rv = line(pkt, item)["row_version"]
|
||
return c.patch(f"/api/packets/{pkt['id']}/bom/{item}", json=_it(fields, it), headers=_im(rv))
|
||
|
||
return call
|
||
|
||
|
||
def short_save(item: int, **body: Any) -> Call:
|
||
def call(c: TestClient, pkt: dict[str, Any], it: str | None = None) -> Any:
|
||
return c.post(f"/api/packets/{pkt['id']}/bom/{item}/short", json=_it(body, it))
|
||
|
||
return call
|
||
|
||
|
||
def short_clear(item: int) -> Call:
|
||
def call(c: TestClient, pkt: dict[str, Any], it: str | None = None) -> Any:
|
||
return c.request("DELETE", f"/api/packets/{pkt['id']}/bom/{item}/short", json=_it({"expect_open": True}, it))
|
||
|
||
return call
|
||
|
||
|
||
def receipt(item: int, **body: Any) -> Call:
|
||
def call(c: TestClient, pkt: dict[str, Any], it: str | None = None) -> Any:
|
||
return c.post(f"/api/packets/{pkt['id']}/bom/{item}/short/receipts", json=_it(body, it))
|
||
|
||
return call
|
||
|
||
|
||
def transition(action: str, **kw: Any) -> Call:
|
||
def call(c: TestClient, pkt: dict[str, Any], it: str | None = None) -> Any:
|
||
return step(c, pkt, action, **_it(kw, it))
|
||
|
||
return call
|
||
|
||
|
||
def refused_then_done(
|
||
shop: Shop,
|
||
pkt: dict[str, Any],
|
||
call: Call,
|
||
performer: str,
|
||
refused: tuple[str, ...] = REFUSED_IN_WH,
|
||
) -> dict[str, Any]:
|
||
"""Every refused role gets 403 forbidden, admin.ann without an IT reason 400 it_reason_required; then the
|
||
performer (wh.wade, or admin.ann with an IT reason) does it. Returns the new read model."""
|
||
for sam in refused:
|
||
r = call(shop(sam), pkt, None)
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden", (sam, r.text)
|
||
r = call(shop("admin.ann"), pkt, None)
|
||
assert r.status_code == 400 and r.json()["code"] == "it_reason_required", r.text
|
||
it = IT if performer == "admin.ann" else None
|
||
r = call(shop(performer), pkt, it)
|
||
assert r.status_code == 200, (performer, r.text)
|
||
out: dict[str, Any] = r.json()
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- every action, every role, the log texts
|
||
|
||
|
||
# (call, log action, log note or None for "do not check", audit action)
|
||
def walk_steps(mtr_id: str) -> list[tuple[Call, str, str | None, str]]:
|
||
return [
|
||
(wh_move("pulling", "incoming"), "Warehouse: Started pulling", None, "packet.set_wh_stage"),
|
||
(
|
||
wh_move("hold", "pulling", reason="Drawing / BOM problem", note="Asked Leo"),
|
||
"Warehouse: Put on hold",
|
||
"Drawing / BOM problem · Asked Leo",
|
||
"packet.set_wh_stage",
|
||
),
|
||
(wh_move("pulling", "hold"), "Warehouse: Pulling resumed", None, "packet.set_wh_stage"),
|
||
(bom_patch(1, heat="HT-1"), "Heat recorded · item 1", "HT-1", "packet.edit_bom"),
|
||
(bom_patch(1, pulled=True), "Pulled · item 1", None, "packet.edit_bom"),
|
||
(bom_patch(1, pulled=False), "Pulled undone · item 1", None, "packet.edit_bom"),
|
||
(bom_patch(1, pulled=True), "Pulled · item 1", None, "packet.edit_bom"),
|
||
(bom_patch(2, na=False), "Traceable again · item 2", None, "packet.edit_bom"),
|
||
(bom_patch(2, na=True), "Marked N/A · item 2", None, "packet.edit_bom"),
|
||
(bom_patch(4, issued=2), "Issued qty changed · item 4", "2 of 4", "packet.edit_bom"),
|
||
(bom_patch(4, loc="Rack 4"), "Location set · item 4", "Rack 4", "packet.edit_bom"),
|
||
(bom_patch(1, heat=None), "Heat cleared · item 1", None, "packet.edit_bom"),
|
||
(bom_patch(1, mtr_asset_id=mtr_id), "MTR attached · item 1", None, "packet.edit_bom"),
|
||
(bom_patch(1, mtr_asset_id=None), "MTR removed · item 1", None, "packet.edit_bom"),
|
||
(bom_patch(1, heat="HT-2", loc="Bin 2"), "Line changed · item 1", None, "packet.edit_bom"),
|
||
(
|
||
short_save(4, qty=2, reason="On order", ref="PO-4471", eta="2026-11-20", expect_open=False),
|
||
"Short recorded · item 4",
|
||
"2 of 4 short · On order · PO-4471 · ETA 2026-11-20",
|
||
"packet.edit_bom",
|
||
),
|
||
(
|
||
short_save(4, qty=1, reason="Back-ordered", expect_open=True),
|
||
"Short updated · item 4",
|
||
"1 of 4 short · Back-ordered",
|
||
"packet.edit_bom",
|
||
),
|
||
(short_clear(4), "Short cleared · item 4", None, "packet.edit_bom"),
|
||
(
|
||
short_save(4, qty=1, reason="Not in stock", heat="HT-4", expect_open=False),
|
||
"Short recorded · item 4",
|
||
"1 of 4 short · Not in stock",
|
||
"packet.edit_bom",
|
||
),
|
||
(bom_patch(3, pulled=True), "Pulled · item 3", None, "packet.edit_bom"),
|
||
(wh_move("pulled", "pulling"), "Warehouse: Marked pulled, ready for production", None, "packet.set_wh_stage"),
|
||
(wh_move("pulling", "pulled"), "Warehouse: Back to pulling", None, "packet.set_wh_stage"),
|
||
(wh_move("pulled", "pulling"), "Warehouse: Marked pulled, ready for production", None, "packet.set_wh_stage"),
|
||
(
|
||
transition("wh_release"),
|
||
"Released to supervisor with 1 short",
|
||
"2 heats recorded · SHORT: item 4 ×1 (Not in stock)",
|
||
"packet.wh_release",
|
||
),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("performer", ["wh.wade", "admin.ann"])
|
||
def test_every_warehouse_action_as_every_role(shop: Shop, performer: str) -> None:
|
||
s = shop.settings
|
||
_, uid = make_unit_sql(s, "L9W" + ("1" if performer == "wh.wade" else "2"))
|
||
pkt = warehouse_packet(shop, uid)
|
||
pid = pkt["id"]
|
||
mtr = upload_mtr(shop("wh.wade"), tiny_png())
|
||
for call, action, note, audit_action in walk_steps(mtr["id"]):
|
||
before = audit_count(s, pid)
|
||
# The refusals are checked once, on the wh.wade walk; the admin walk checks the IT reason only.
|
||
pkt = refused_then_done(shop, pkt, call, performer, REFUSED_IN_WH if performer == "wh.wade" else ())
|
||
lg = last_log(shop("wh.wade"), pid)
|
||
assert lg["action"] == action, (action, lg)
|
||
if note is not None:
|
||
assert lg["note"] == note, (action, lg)
|
||
if action.startswith("Heat recorded"):
|
||
assert lg["note"] == "HT-1"
|
||
assert audit_count(s, pid) == before + 1, action
|
||
au = last_audit(s, pid)
|
||
assert au[0] == audit_action, (action, au)
|
||
if performer == "admin.ann":
|
||
assert lg["it_reason"] == IT and lg["by"]["display_name"] == "Ann Admin", lg
|
||
assert au[2] == IT
|
||
else:
|
||
assert lg["it_reason"] is None and lg["by"]["display_name"] == "Wade Warehouse"
|
||
if audit_action == "packet.edit_bom" and "item" in action:
|
||
assert int(au[1]["item"]) == int(action.rsplit(" ", 1)[1])
|
||
if action.startswith("Short recorded") or action.startswith("Short updated"):
|
||
assert float(au[1]["qty"]) in (1.0, 2.0), au
|
||
if audit_action == "packet.set_wh_stage":
|
||
assert au[1]["wh_from"] and au[1]["wh_to"]
|
||
|
||
assert pkt["status"] == "ready" and pkt["open_shorts"] == 1
|
||
assert line(pkt, 1)["heat"] == "HT-2" and line(pkt, 1)["loc"] == "Bin 2" and line(pkt, 1)["mtr_asset_id"] is None
|
||
assert line(pkt, 2)["na"] is True and line(pkt, 4)["heat"] == "HT-4"
|
||
# The short receipt on the ready packet: the warehouse (or IT) receives; Clerical, Q.C., PM and a welder cannot.
|
||
pkt = refused_then_done(shop, pkt, receipt(4, qty=1, heat="HT-5"), performer, REFUSED_IN_READY)
|
||
lg = last_log(shop("wh.wade"), pid)
|
||
assert lg["action"] == "Short received · item 4"
|
||
assert lg["note"] == "1 pc · heat HT-5 · short closed · packet ready to assign"
|
||
au = last_audit(s, pid)
|
||
assert float(au[1]["qty"]) == 1 and int(au[1]["item"]) == 4
|
||
assert au[0] == ("packet.wh_receive_short" if performer == "wh.wade" else au[0])
|
||
assert pkt["open_shorts"] == 0 and line(pkt, 4)["short"] is None
|
||
|
||
|
||
@pytest.mark.parametrize("performer", ["wh.wade", "admin.ann"])
|
||
def test_send_back_to_clerical(shop: Shop, performer: str) -> None:
|
||
_, uid = make_unit_sql(shop.settings, "L9S" + ("1" if performer == "wh.wade" else "2"))
|
||
pkt = warehouse_packet(shop, uid)
|
||
reason = "The BOM is the old revision"
|
||
short = shop("wh.wade").post(f"/api/packets/{pkt['id']}/transitions", json={"action": "wh_send_back"})
|
||
assert short.status_code in (400, 422), short.text
|
||
r = step(shop("wh.wade"), pkt, "wh_send_back", reason="too few")
|
||
assert r.status_code == 400 and r.json()["code"] == "reason_required", r.text
|
||
pkt = refused_then_done(shop, pkt, transition("wh_send_back", reason=reason), performer)
|
||
assert pkt["status"] == "draft"
|
||
lg = last_log(shop("clerk.clara"), pkt["id"])
|
||
assert lg["action"] == "Sent back to Clerical" and lg["note"] == reason
|
||
assert last_audit(shop.settings, pkt["id"])[0] == "packet.wh_send_back"
|
||
|
||
|
||
# ---------------------------------------------------------------- hold reasons and stage moves
|
||
|
||
|
||
def test_hold_reasons_and_stage_moves(shop: Shop) -> None:
|
||
_, uid = make_unit_sql(shop.settings, "L9H1")
|
||
pkt = warehouse_packet(shop, uid)
|
||
c = shop("wh.wade")
|
||
url = f"/api/packets/{pkt['id']}/wh"
|
||
assert c.post(url, json={"stage": "pulling", "expect_stage": "incoming"}).status_code == 200
|
||
same = c.post(url, json={"stage": "pulling", "expect_stage": "pulling"})
|
||
assert same.status_code == 409 and same.json()["code"] == "wh_transition", same.text
|
||
stale = c.post(url, json={"stage": "hold", "reason": "Other", "note": "x", "expect_stage": "incoming"})
|
||
assert stale.status_code == 409 and stale.json()["code"] == "stale_step", stale.text
|
||
assert stale.json()["current"]["stage"] == "pulling"
|
||
bad = c.post(url, json={"stage": "hold", "reason": "Lunch break", "expect_stage": "pulling"})
|
||
assert bad.status_code == 400 and bad.json()["code"] == "invalid_reason", bad.text
|
||
assert bad.json()["detail"] == "Pick one of the hold reasons."
|
||
other = c.post(url, json={"stage": "hold", "reason": "Other", "expect_stage": "pulling"})
|
||
assert other.status_code == 400 and other.json()["code"] == "reason_required", other.text
|
||
assert other.json()["detail"] == "Say what it's waiting on."
|
||
none = c.post(url, json={"stage": "hold", "expect_stage": "pulling"})
|
||
assert none.status_code == 400, none.text
|
||
for reason in HOLD_REASONS:
|
||
body: dict[str, Any] = {"stage": "hold", "reason": reason, "expect_stage": "pulling"}
|
||
if reason == "Other":
|
||
body["note"] = "Crane is down"
|
||
r = c.post(url, json=body)
|
||
assert r.status_code == 200, (reason, r.text)
|
||
assert r.json()["wh"]["stage"] == "hold" and r.json()["wh"]["reason"] == reason
|
||
lg = last_log(c, pkt["id"])
|
||
assert lg["action"] == "Warehouse: Put on hold"
|
||
assert lg["note"] == (f"{reason} · Crane is down" if reason == "Other" else reason)
|
||
r = c.post(url, json={"stage": "pulling", "expect_stage": "hold"})
|
||
assert r.status_code == 200 and last_log(c, pkt["id"])["action"] == "Warehouse: Pulling resumed"
|
||
long = c.post(url, json={"stage": "hold", "reason": "Other", "note": "n" * 300, "expect_stage": "pulling"})
|
||
assert long.status_code == 200, long.text
|
||
assert len(long.json()["wh"]["note"]) == 200
|
||
c.post(url, json={"stage": "pulling", "expect_stage": "hold"})
|
||
r = c.post(url, json={"stage": "pulled", "expect_stage": "pulling"})
|
||
assert r.status_code == 200 and r.json()["wh"]["stage"] == "pulled"
|
||
r = c.post(url, json={"stage": "pulling", "expect_stage": "pulled"})
|
||
assert r.status_code == 200 and r.json()["wh"]["stage"] == "pulling"
|
||
assert last_log(c, pkt["id"])["action"] == "Warehouse: Back to pulling"
|
||
bad = c.post(url, json={"stage": "incoming", "expect_stage": "pulling"})
|
||
assert bad.status_code in (409, 422), bad.text
|
||
|
||
|
||
def test_pulled_tick_auto_starts(shop: Shop) -> None:
|
||
s = shop.settings
|
||
_, uid = make_unit_sql(s, "L9A1")
|
||
c = shop("wh.wade")
|
||
# From incoming
|
||
a = warehouse_packet(shop, uid)
|
||
r = bom_patch(1, pulled=True)(c, a, None)
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["wh"]["stage"] == "pulling" and r.json()["wh"]["auto"] is True
|
||
lg = last_log(c, a["id"])
|
||
assert lg["action"] == "Pulled · item 1" and lg["note"].endswith("started pulling"), lg
|
||
row = _one(s, "SELECT stage, auto FROM packet_wh WHERE packet_id = %s AND closed_at IS NULL", (a["id"],))
|
||
assert row == ("pulling", True)
|
||
# A second tick does not move the stage again
|
||
r = bom_patch(4, pulled=True)(c, r.json(), None)
|
||
assert last_log(c, a["id"])["action"] == "Pulled · item 4"
|
||
assert int(_one(s, "SELECT count(*) FROM packet_wh WHERE packet_id = %s", (a["id"],))[0]) == 2
|
||
# From hold
|
||
_, uid2 = make_unit_sql(s, "L9A2")
|
||
b = warehouse_packet(shop, uid2)
|
||
r = c.post(
|
||
f"/api/packets/{b['id']}/wh",
|
||
json={"stage": "hold", "reason": "Waiting on purchasing / PO", "expect_stage": "incoming"},
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
r = bom_patch(1, pulled=True)(c, r.json(), None)
|
||
assert r.status_code == 200 and r.json()["wh"]["stage"] == "pulling" and r.json()["wh"]["auto"] is True
|
||
lg = last_log(c, b["id"])
|
||
assert lg["action"] == "Pulled · item 1" and lg["note"].endswith("pulling resumed"), lg
|
||
row = _one(s, "SELECT stage, auto FROM packet_wh WHERE packet_id = %s AND closed_at IS NULL", (b["id"],))
|
||
assert row == ("pulling", True)
|
||
|
||
|
||
# ---------------------------------------------------------------- heats and MTRs
|
||
|
||
|
||
BAD_HEATS = ("<img src=x>", "A" * 41, " HT1", "HT@1", "-HT1")
|
||
|
||
|
||
def test_bad_heat_is_refused(shop: Shop) -> None:
|
||
_, uid = make_unit_sql(shop.settings, "L9B1")
|
||
pkt = warehouse_packet(shop, uid)
|
||
for value in BAD_HEATS:
|
||
r = bom_patch(1, heat=value)(shop("wh.wade"), pkt, None)
|
||
assert r.status_code == 422 and r.json()["code"] == "invalid_heat", (value, r.text)
|
||
if value in ("<img src=x>", "HT@1"):
|
||
assert r.json()["detail"] == (
|
||
f'Heat number "{value}" isn\'t valid: letters, numbers, space . / # _ - only, up to 40 characters.'
|
||
)
|
||
assert fresh(shop("wh.wade"), pkt["id"])["bom"][0]["heat"] is None
|
||
|
||
|
||
def test_good_heats_are_recorded(shop: Shop) -> None:
|
||
_, uid = make_unit_sql(shop.settings, "L9G1")
|
||
pkt = warehouse_packet(shop, uid)
|
||
c = shop("wh.wade")
|
||
for value in ("HT-1234/7A", "7A", "A 1.2#3_x", "A" * 40):
|
||
r = bom_patch(1, heat=value)(c, pkt, None)
|
||
assert r.status_code == 200, (value, r.text)
|
||
pkt = r.json()
|
||
assert line(pkt, 1)["heat"] == value
|
||
lg = last_log(c, pkt["id"])
|
||
assert lg["action"] == "Heat recorded · item 1" and lg["note"] == value
|
||
# A trailing space from a scanner is trimmed
|
||
r = bom_patch(1, heat="HT-9 ")(c, pkt, None)
|
||
assert r.status_code == 200 and line(r.json(), 1)["heat"] == "HT-9"
|
||
|
||
|
||
def test_mtr_must_be_a_pdf_or_image(shop: Shop) -> None:
|
||
_, uid = make_unit_sql(shop.settings, "L9M1")
|
||
pkt = warehouse_packet(shop, uid)
|
||
c = shop("wh.wade")
|
||
missing = bom_patch(1, mtr_asset_id="6f1c1d1e-0000-4000-8000-000000000000")(c, pkt, None)
|
||
assert missing.status_code == 422 and missing.json()["code"] == "invalid_asset"
|
||
csv = shop("clerk.clara").post(
|
||
"/api/assets", files={"file": ("bom.csv", b"item,qty\n1,4\n", "text/csv")}, data={"purpose": "bom"}
|
||
)
|
||
assert csv.status_code == 201, csv.text
|
||
wrong = bom_patch(1, mtr_asset_id=csv.json()["id"])(c, pkt, None)
|
||
assert wrong.status_code == 422 and wrong.json()["code"] == "invalid_asset", wrong.text
|
||
assert wrong.json()["detail"] == "An MTR is a PDF, PNG or JPG file."
|
||
png = upload_mtr(c, tiny_png())
|
||
r = bom_patch(1, mtr_asset_id=png["id"])(c, pkt, None)
|
||
assert r.status_code == 200 and line(r.json(), 1)["mtr_asset_id"] == png["id"], r.text
|
||
r = bom_patch(1, mtr_asset_id=None)(c, r.json(), None)
|
||
assert r.status_code == 200 and line(r.json(), 1)["mtr_asset_id"] is None
|
||
|
||
|
||
# ---------------------------------------------------------------- the release gate
|
||
|
||
|
||
def test_release_gate(shop: Shop) -> None:
|
||
s = shop.settings
|
||
_, uid = make_unit_sql(s, "L9R1")
|
||
pkt = warehouse_packet(shop, uid)
|
||
pid = pkt["id"]
|
||
c = shop("wh.wade")
|
||
pkt = bom_patch(1, heat="HT-1")(c, pkt, None).json()
|
||
pkt = bom_patch(1, pulled=True)(c, pkt, None).json()
|
||
pkt = bom_patch(3, pulled=True)(c, pkt, None).json() # the sub line needs no heat
|
||
# Item 4 is not pulled: the first rule only (the N/A line and the sub line need nothing)
|
||
assert "wh_release" not in pkt["can"] and pkt["blocked"]["wh_release"] == [MSG_PULL], pkt["blocked"]
|
||
r = step(c, pkt, "wh_release")
|
||
assert r.status_code == 403 and r.json()["code"] == "precondition" and r.json()["detail"] == MSG_PULL
|
||
# Every line pulled, item 4 has no heat
|
||
pkt = bom_patch(4, pulled=True)(c, pkt, None).json()
|
||
assert pkt["blocked"]["wh_release"] == [MSG_HEAT], pkt["blocked"]
|
||
r = step(c, pkt, "wh_release")
|
||
assert r.status_code == 403 and r.json()["code"] == "precondition" and r.json()["detail"] == MSG_HEAT
|
||
card = next(x for x in c.get("/api/packets", params={"status": "warehouse"}).json()["packets"] if x["id"] == pid)
|
||
assert card["bom_total"] == 3 and card["bom_pulled"] == 3 and card["bom_missing_heat"] == 1
|
||
# Unpull item 4 and record a short instead: the first rule passes, the heat of the pulled pieces is missing
|
||
pkt = bom_patch(4, pulled=False)(c, pkt, None).json()
|
||
assert pkt["blocked"]["wh_release"] == [MSG_PULL]
|
||
r = short_save(4, qty=1, reason="Not in stock", expect_open=False)(c, pkt, None)
|
||
assert r.status_code == 200, r.text
|
||
pkt = r.json()
|
||
assert pkt["blocked"]["wh_release"] == [MSG_HEAT], pkt["blocked"]
|
||
r = short_save(4, qty=1, reason="Not in stock", heat="HT-4", expect_open=True)(c, pkt, None)
|
||
assert r.status_code == 200, r.text
|
||
pkt = r.json()
|
||
assert "wh_release" in pkt["can"] and "wh_release" not in pkt["blocked"]
|
||
pkt = ok_step(c, pkt, "wh_release")
|
||
assert pkt["status"] == "ready"
|
||
lg = last_log(c, pid)
|
||
assert lg["action"] == "Released to supervisor with 1 short"
|
||
assert lg["note"] == "2 heats recorded · SHORT: item 4 ×1 (Not in stock)"
|
||
board = c.get("/api/warehouse/board").json()
|
||
rel = [x for x in board["columns"]["released"] if x["id"] == pid]
|
||
assert len(rel) == 1 and rel[0]["released_at"], board["columns"]["released"]
|
||
assert board["kpis"]["released_24h"] >= 1
|
||
assert all(pid not in [x["id"] for x in board["columns"][k]] for k in ("incoming", "pulling", "hold", "pulled"))
|
||
# Sent back and released again: it shows once in the released column
|
||
pkt = ok_step(shop("lead.lou"), pkt, "send_back_to_warehouse", reason="Wrong studs were pulled")
|
||
assert pkt["status"] == "warehouse"
|
||
pkt = ok_step(c, fresh(c, pid), "wh_release")
|
||
lg = last_log(c, pid)
|
||
assert lg["action"] == "Released to supervisor with 1 short"
|
||
board = c.get("/api/warehouse/board").json()
|
||
assert [x["id"] for x in board["columns"]["released"]].count(pid) == 1
|
||
# Without shorts the label has no count
|
||
_, uid2 = make_unit_sql(s, "L9R2")
|
||
other = warehouse_packet(shop, uid2)
|
||
for item in (1, 3, 4):
|
||
other = bom_patch(item, pulled=True)(c, other, None).json()
|
||
for item in (1, 4):
|
||
other = bom_patch(item, heat=f"HT-{item}0")(c, other, None).json()
|
||
other = ok_step(c, other, "wh_release")
|
||
lg = last_log(c, other["id"])
|
||
assert lg["action"] == "Released to supervisor" and lg["note"] == "2 heats recorded"
|
||
|
||
|
||
def test_release_gate_without_traceable_lines(shop: Shop) -> None:
|
||
"""A packet whose lines are all N/A or sub-assemblies needs no heat at all."""
|
||
_, uid = make_unit_sql(shop.settings, "L9R3")
|
||
pkt = warehouse_packet(shop, uid)
|
||
c = shop("wh.wade")
|
||
pkt = bom_patch(1, na=True)(c, pkt, None).json()
|
||
pkt = bom_patch(4, na=True)(c, pkt, None).json()
|
||
assert pkt["blocked"]["wh_release"] == [MSG_PULL]
|
||
pkt = bom_patch(3, pulled=True)(c, pkt, None).json()
|
||
assert "wh_release" in pkt["can"], pkt["blocked"]
|
||
|
||
|
||
# ---------------------------------------------------------------- receiving a short by custody
|
||
|
||
|
||
def test_receive_short_by_custody(shop: Shop) -> None:
|
||
s = shop.settings
|
||
_, uid = make_unit_sql(s, "L9C1")
|
||
pkt = warehouse_packet(shop, uid)
|
||
pid = pkt["id"]
|
||
wade = shop("wh.wade")
|
||
pkt = bom_patch(1, heat="HT-1")(wade, pkt, None).json()
|
||
pkt = bom_patch(1, pulled=True)(wade, pkt, None).json()
|
||
pkt = bom_patch(3, pulled=True)(wade, pkt, None).json()
|
||
pkt = short_save(4, qty=3, reason="On order", ref="PO-88", heat="HT-4", expect_open=False)(wade, pkt, None).json()
|
||
pkt = ok_step(wade, pkt, "wh_release")
|
||
assert pkt["status"] == "ready" and pkt["open_shorts"] == 1
|
||
|
||
r = receipt(4, qty=1, heat="HT-R1")(wade, pkt, None)
|
||
assert r.status_code == 200, r.text
|
||
pkt = r.json()
|
||
lg = last_log(wade, pid)
|
||
assert lg["action"] == "Short received · item 4"
|
||
assert lg["note"] == "1 pc · heat HT-R1 · 2 still short · packet ready to assign"
|
||
assert last_audit(s, pid)[0] == "packet.wh_receive_short"
|
||
# The warehouse does not hold the ready packet: no other line edit
|
||
r = bom_patch(1, heat="HT-X")(wade, pkt, None)
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden", r.text
|
||
r = bom_patch(1, loc="Rack 9")(wade, pkt, None)
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden", r.text
|
||
|
||
for sam in ("qc.quinn", "clerk.clara", "pm.paula", "weld.andy"):
|
||
r = receipt(4, qty=1, heat="HT-R2")(shop(sam), pkt, None)
|
||
assert r.status_code == 403 and r.json()["code"] == "forbidden", (sam, r.text)
|
||
lou = shop("lead.lou")
|
||
over = receipt(4, qty=3, heat="HT-R2")(lou, pkt, None)
|
||
assert over.status_code == 422, over.text
|
||
noheat = receipt(4, qty=1)(lou, pkt, None)
|
||
assert noheat.status_code == 422 and noheat.json()["code"] == "heat_required", noheat.text
|
||
r = receipt(4, qty=1, heat="HT-R2")(lou, pkt, None)
|
||
assert r.status_code == 200, r.text
|
||
pkt = r.json()
|
||
assert last_log(lou, pid)["by"]["display_name"] == "Lou Lead"
|
||
assert last_audit(s, pid)[0] == "packet.edit_bom"
|
||
r = receipt(4, qty=1, heat="HT-R3")(shop("sup.wes"), pkt, None)
|
||
assert r.status_code == 200, r.text
|
||
pkt = r.json()
|
||
assert pkt["open_shorts"] == 0 and line(pkt, 4)["short"] is None
|
||
assert line(pkt, 4)["heat"] == "HT-4 / HT-R1 / HT-R2 / HT-R3"
|
||
assert last_log(wade, pid)["note"] == "1 pc · heat HT-R3 · short closed · packet ready to assign"
|
||
closed = _one(
|
||
s,
|
||
"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 = 4",
|
||
(pid,),
|
||
)
|
||
assert closed[0] is not None and float(closed[1]) == 0
|
||
r = receipt(4, qty=1, heat="HT-R4")(wade, pkt, None)
|
||
assert r.status_code in (403, 404), r.text # nothing is short now
|
||
|
||
# The holder may open a short on a ready packet; a receipt on an N/A line needs no heat.
|
||
r = short_save(2, qty=2, reason="Damaged / rejected", expect_open=False)(lou, pkt, None)
|
||
assert r.status_code == 200, r.text
|
||
pkt = r.json()
|
||
with app_conn(s) as conn:
|
||
stall = station_id(conn)
|
||
andy = user_id(conn, "weld.andy")
|
||
pkt = ok_step(shop("sup.wes"), pkt, "assign", station_id=stall, welder_id=andy)
|
||
r = receipt(2, qty=2)(wade, pkt, None)
|
||
assert r.status_code == 200, r.text
|
||
lg = last_log(wade, pid)
|
||
assert lg["action"] == "Short received · item 2"
|
||
assert lg["note"] == "2 pc · short closed · packet at a station in Stall 1"
|
||
|
||
|
||
def test_short_reasons_are_a_closed_set(shop: Shop) -> None:
|
||
_, uid = make_unit_sql(shop.settings, "L9Q1")
|
||
pkt = warehouse_packet(shop, uid)
|
||
c = shop("wh.wade")
|
||
bad = short_save(4, qty=1, reason="Lost it", expect_open=False)(c, pkt, None)
|
||
assert bad.status_code == 400 and bad.json()["code"] == "invalid_reason", bad.text
|
||
assert bad.json()["detail"] == "Pick one of the short reasons."
|
||
for q in (0, 5):
|
||
r = short_save(4, qty=q, reason="Other", note="x", expect_open=False)(c, pkt, None)
|
||
assert r.status_code == 422, (q, r.text)
|
||
reasons = (
|
||
"Not in stock",
|
||
"On order",
|
||
"Back-ordered",
|
||
"Damaged / rejected",
|
||
"Wrong part received",
|
||
"Needs cutting / prep",
|
||
"Other",
|
||
)
|
||
open_ = False
|
||
for reason in reasons:
|
||
r = short_save(4, qty=1, reason=reason, note="see PO", expect_open=open_)(c, pkt, None)
|
||
assert r.status_code == 200, (reason, r.text)
|
||
open_ = True
|
||
assert line(r.json(), 4)["short"]["reason"] == reason
|
||
|
||
|
||
# ---------------------------------------------------------------- the board endpoint
|
||
|
||
|
||
def test_board(shop: Shop) -> None:
|
||
s = shop.settings
|
||
proj_a, unit_a = make_unit_sql(s, "L9K1")
|
||
proj_b, unit_b = make_unit_sql(s, "L9K2")
|
||
p1 = warehouse_packet(shop, unit_a, drawing="D-OLD")
|
||
p2 = warehouse_packet(shop, unit_a, drawing="D-NEW")
|
||
p5 = warehouse_packet(shop, unit_a, drawing="D-REL")
|
||
p3 = warehouse_packet(shop, unit_a, drawing="D-HOLD")
|
||
p4 = warehouse_packet(shop, unit_b, drawing="D-OTHER")
|
||
with app_conn(s) as conn:
|
||
conn.execute("UPDATE packets SET wh_sent_at = now() - interval '3 days' WHERE id = %s", (p1["id"],))
|
||
conn.execute(
|
||
"UPDATE packets SET wh_sent_at = date_trunc('day', now()) - interval '1 day' + interval '9 hours'"
|
||
" WHERE id = %s",
|
||
(p2["id"],),
|
||
)
|
||
conn.execute(
|
||
"UPDATE packets SET wh_sent_at = date_trunc('day', now()) - interval '1 day' + interval '10 hours',"
|
||
" visual_rel = '2026-11-01' WHERE id = %s",
|
||
(p5["id"],),
|
||
)
|
||
wade = shop("wh.wade")
|
||
r = wade.post(
|
||
f"/api/packets/{p3['id']}/wh",
|
||
json={"stage": "hold", "reason": "Drawing / BOM problem", "note": "Asked Leo", "expect_stage": "incoming"},
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
r = wade.post(f"/api/packets/{p4['id']}/wh", json={"stage": "pulling", "expect_stage": "incoming"})
|
||
assert r.status_code == 200, r.text
|
||
p4 = r.json()
|
||
p4 = short_save(4, qty=1, reason="On order", ref="PO-1", eta="2020-01-02", expect_open=False)(wade, p4, None).json()
|
||
p1 = short_save(1, qty=1, reason="Not in stock", expect_open=False)(wade, p1, None).json()
|
||
assert p1["wh"]["stage"] == "pulling" # the short auto-started it
|
||
|
||
r = shop("pm.paula").get("/api/warehouse/board")
|
||
assert r.status_code == 200, r.text
|
||
b = r.json()
|
||
assert set(b) >= {"kpis", "columns", "shorts"}
|
||
assert b["kpis"] == {"waiting": 2, "pulling": 2, "hold": 1, "pulled": 0, "open_shorts": 2, "released_24h": 0}
|
||
cols = {k: [x["id"] for x in v] for k, v in b["columns"].items()}
|
||
assert set(cols) == {"incoming", "pulling", "hold", "pulled", "released"}
|
||
assert cols["incoming"] == [p5["id"], p2["id"]] # same day: the earlier VISUAL release date first
|
||
assert cols["pulling"] == [p1["id"], p4["id"]] # the older day first
|
||
assert cols["hold"] == [p3["id"]] and cols["pulled"] == [] and cols["released"] == []
|
||
hold = b["columns"]["hold"][0]
|
||
assert hold["wh"]["stage"] == "hold" and hold["wh"]["reason"] == "Drawing / BOM problem"
|
||
assert hold["wh"]["note"] == "Asked Leo" and hold["wh"]["auto"] is False
|
||
assert hold["wh"]["by"]["display_name"] == "Wade Warehouse" and hold["wh"]["at"]
|
||
inc = b["columns"]["incoming"][0]
|
||
assert inc["visual_rel"] == "2026-11-01" and inc["wh_sent_at"]
|
||
assert inc["bom_total"] == 3 and inc["bom_pulled"] == 0 and inc["bom_missing_heat"] == 0
|
||
|
||
shorts = b["shorts"]
|
||
assert [(x["packet"]["id"], x["item"]) for x in shorts] == [(p4["id"], 4), (p1["id"], 1)]
|
||
late, plain = shorts
|
||
assert late["late"] is True and plain["late"] is False
|
||
assert float(late["qty"]) == 4 and float(late["qty_open"]) == 1 and late["reason"] == "On order"
|
||
assert late["ref"] == "PO-1" and late["eta"] == "2020-01-02" and late["age_days"] == 0
|
||
assert late["part_id"] == "STUD-58" and late["description"] == "5/8 stud bolt"
|
||
assert late["short_id"] and late["row_version"] >= 1 and late["bom_line_id"] and late["opened_at"]
|
||
pk = late["packet"]
|
||
assert pk["status"] == "warehouse" and pk["wh_stage"] == "pulling" and pk["holder"] == "Warehouse"
|
||
assert pk["dept"]["code"] == "03" and pk["station"] is None and pk["drawing"] == "D-OTHER"
|
||
assert plain["eta"] is None and plain["material"] == "A106-B" and plain["length"] == "20ft"
|
||
|
||
def board(**params: Any) -> dict[str, Any]:
|
||
r = wade.get("/api/warehouse/board", params=params)
|
||
assert r.status_code == 200, r.text
|
||
out: dict[str, Any] = r.json()
|
||
return out
|
||
|
||
def ids(bd: dict[str, Any]) -> set[int]:
|
||
return {x["id"] for v in bd["columns"].values() for x in v}
|
||
|
||
by_project = board(project_id=proj_b)
|
||
assert ids(by_project) == {p4["id"]} and [x["packet"]["id"] for x in by_project["shorts"]] == [p4["id"]]
|
||
assert by_project["kpis"]["waiting"] == 0 and by_project["kpis"]["pulling"] == 1
|
||
assert by_project["kpis"]["hold"] == 0 and by_project["kpis"]["open_shorts"] == 1
|
||
by_unit = board(unit_id=unit_a)
|
||
assert ids(by_unit) == {p1["id"], p2["id"], p3["id"], p5["id"]}
|
||
assert by_unit["kpis"]["waiting"] == 2 and by_unit["kpis"]["pulling"] == 1 and by_unit["kpis"]["hold"] == 1
|
||
by_q = board(q="d-hold")
|
||
assert ids(by_q) == {p3["id"]} and by_q["shorts"] == []
|
||
assert by_q["kpis"] == b["kpis"] # the KPIs follow the project and unit only
|
||
by_dept = board(dept="06")
|
||
assert ids(by_dept) == set() and by_dept["shorts"] == [] and by_dept["kpis"] == b["kpis"]
|
||
assert ids(board(dept="03")) == ids(b)
|
||
assert proj_a != proj_b
|
||
|
||
# Every card of the packet list carries the warehouse fields
|
||
cards = {x["id"]: x for x in wade.get("/api/packets", params={"status": "warehouse"}).json()["packets"]}
|
||
c4 = cards[p4["id"]]
|
||
assert c4["wh"]["stage"] == "pulling" and c4["wh_stage"] == "pulling"
|
||
assert c4["bom_total"] == 3 and c4["bom_pulled"] == 1 and c4["bom_missing_heat"] == 1
|
||
assert "wh_sent_at" in c4 and "visual_rel" in c4
|
||
# A removed packet's shorts are not listed
|
||
with app_conn(s) as conn:
|
||
conn.execute("UPDATE packets SET removed_at = now() WHERE id = %s", (p4["id"],))
|
||
after = board()
|
||
assert p4["id"] not in [x["packet"]["id"] for x in after["shorts"]]
|
||
assert p4["id"] not in ids(after)
|
||
|
||
|
||
def test_board_needs_a_session(shop: Shop) -> None:
|
||
as_user(shop.main, "weld.andy")
|
||
r = shop.main.get("/api/warehouse/board")
|
||
assert r.status_code == 200 and r.json()["kpis"]["waiting"] == 0
|
||
anon = TestClient(shop.main.app, headers=ORIGIN_HEADERS)
|
||
assert anon.get("/api/warehouse/board").status_code == 401
|