"""M3 smoke test — run creation, list/detail, quantity override (docs/30 §D.2, FR-MFG-08/09/18). Still touches no stock: creating a run only copies and scales the template. Depends on the template m2_templates.py leaves behind, so run that first: python Backend/smoke/m2_templates.py python Backend/smoke/m3_runs.py """ from __future__ import annotations import sys from smoke_common import bootstrap TEMPLATE_CODE = "SMOKE-PT-M2" TARGET_QTY = 50 def find_template(c): listed = c.get(f"/production-templates?q={TEMPLATE_CODE}") row = next((t for t in listed.body["items"] if t["code"] == TEMPLATE_CODE), None) if not row: sys.exit(f"FATAL: template {TEMPLATE_CODE} not found — run m2_templates.py first.") return row["templateId"] def main(): c, chk, args = bootstrap(__doc__) print(f"API {args.api}") tid = find_template(c) tpl = c.get(f"/production-templates/{tid}").body warehouse = c.get("/warehouses?pageSize=1").body["items"][0]["warehouseId"] # Captured before creating so the script stays re-runnable — a previous run of this # script leaves its own InProgress run behind for M4. runs_before = next(t for t in c.get(f"/production-templates?q={TEMPLATE_CODE}").body["items"] if t["templateId"] == tid)["activeRunCount"] print(f"template={tid} warehouse={warehouse} targetQty={TARGET_QTY} activeRunsBefore={runs_before}") terminal_stage = next(s for s in tpl["stages"] if s["name"] == "Assemble & QA") terminal_qpb = terminal_stage["outputs"][0]["qtyPerBatch"] expected_scale = TARGET_QTY / terminal_qpb # ------------------------------------------------------------------ create chk.section("1. Create a run (FR-MFG-08: scale, copy, number)") created = c.post("/production-runs", { "templateId": tid, "targetQty": TARGET_QTY, "warehouseId": warehouse, }) if not chk.status("POST /production-runs", created, 201): return chk.finish("M3") run = created.body rid = run["runId"] chk.check("ETag header present", created.etag is not None, True) chk.check("docNo uses the PRD sequence", run["docNo"].startswith("PRD-"), True) chk.check("run starts InProgress", run["status"], "InProgress") chk.check("reworkCount starts at 0", run["reworkCount"], 0) chk.check("scaleFactor computed", float(run["scaleFactor"]), float(expected_scale)) chk.check("3 stages copied", len(run["stages"]), 3) chk.check("2 edges copied", len(run["edges"]), 2) # ------------------------------------------------------- copy-on-create chk.section("2. Copy-on-create carries display fields and fieldDefs (FR-MFG-06)") stages = {s["name"]: s for s in run["stages"]} chk.check("stage names copied", sorted(stages), ["Assemble & QA", "Cut frame", "Prep cushions"]) cut = stages["Cut frame"] chk.check("roleLabel copied", cut["roleLabel"], "Carpentry") chk.check("estimatedMinutes copied", cut["estimatedMinutes"], 60) chk.check("posX copied", float(cut["posX"]), 80.0) chk.check("fieldDefs copied", cut["fieldDefs"][0]["key"], "moisture_ok") chk.check("fieldValues empty before complete", cut["fieldValues"], None) chk.check("templateStageId links back", cut["templateStageId"] is not None, True) # -------------------------------------------------------------- readiness chk.section("3. Entry stages Ready, others Waiting (FR-MFG-09)") chk.check("Cut frame is an entry", cut["isEntry"], True) chk.check("Cut frame Ready", cut["status"], "Ready") chk.check("Prep cushions Ready", stages["Prep cushions"]["status"], "Ready") asm = stages["Assemble & QA"] chk.check("Assemble is terminal", asm["isTerminal"], True) chk.check("Assemble is not an entry", asm["isEntry"], False) chk.check("Assemble Waiting", asm["status"], "Waiting") chk.check("no stage is terminal but Cut/Prep", [s["name"] for s in run["stages"] if s["isTerminal"]], ["Assemble & QA"]) # ----------------------------------------------------------------- scaling chk.section("4. Every quantity scaled by the factor (FR-MFG-08)") tpl_stages = {s["name"]: s for s in tpl["stages"]} ok = True for name, rs in stages.items(): ts = tpl_stages[name] for ti, ri in zip(ts["inputs"], rs["inputs"]): want = round(float(ti["qtyPerBatch"]) * expected_scale, 4) if float(ri["plannedQty"]) != want: ok = False print(f" input mismatch on {name}: {ri['plannedQty']} != {want}") for to, ro in zip(ts["outputs"], rs["outputs"]): want = round(float(to["qtyPerBatch"]) * expected_scale, 4) if float(ro["plannedQty"]) != want: ok = False print(f" output mismatch on {name}: {ro['plannedQty']} != {want}") chk.check("all planned quantities == qtyPerBatch x scaleFactor", ok, True) chk.check("Cut frame input scaled (8 x 50)", float(cut["inputs"][0]["plannedQty"]), 400.0) chk.check("terminal output scaled to the target", float(asm["outputs"][0]["plannedQty"]), float(TARGET_QTY)) chk.check("consumed/delivered start at zero", all(float(i["consumedQty"]) == 0 and float(i["deliveredQty"]) == 0 for s in run["stages"] for i in s["inputs"]), True) chk.check("cost pool starts empty", float(run["costPool"]["net"]), 0.0) # Upstream inputs must point at the run's own copied outputs, not the template's. run_output_ids = {o["runOutputId"] for s in run["stages"] for o in s["outputs"]} chk.check("upstream inputs rewired to run outputs", all(i["fromRunOutputId"] in run_output_ids for i in asm["inputs"] if i["source"] == "Upstream"), True) # -------------------------------------------------------------------- list chk.section("5. Run board projection (FR-MFG-18)") listed = c.get(f"/production-runs?q={run['docNo']}") chk.status("GET /production-runs", listed, 200) row = next((r for r in listed.body["items"] if r["runId"] == rid), None) chk.check("run appears on the board", row is not None, True) if row: chk.check("stageSummary counts match", row["stageSummary"], {"waiting": 1, "ready": 2, "inProgress": 0, "done": 0, "approved": 0}) chk.check("templateName joined", row["templateName"], tpl["name"]) chk.check("finished item surfaced", row["finishedItemId"], asm["outputs"][0]["itemId"]) chk.check("finished item name joined", row["finishedItemName"] is not None, True) chk.check("filter by status=InProgress finds it", any(r["runId"] == rid for r in c.get("/production-runs?status=InProgress&pageSize=200").body["items"]), True) chk.check("filter by status=Completed excludes it", any(r["runId"] == rid for r in c.get("/production-runs?status=Completed&pageSize=200").body["items"]), False) chk.check("filter by templateId finds it", any(r["runId"] == rid for r in c.get(f"/production-runs?templateId={tid}&pageSize=200").body["items"]), True) # -------------------------------------------------------------- edit-lock chk.section("6. Template edit-lock now that a run is InProgress (FR-MFG-06)") head = c.get(f"/production-templates/{tid}") locked = c.put(f"/production-templates/{tid}", { "code": tpl["code"], "name": tpl["name"], "description": tpl.get("description"), "stages": [ {"key": s["key"], "name": s["name"], "roleLabel": s.get("roleLabel"), "estimatedMinutes": s["estimatedMinutes"], "posX": s["posX"], "posY": s["posY"], "fieldDefs": s["fieldDefs"], "inputs": [{"source": i["source"], "itemId": i.get("itemId"), "fromOutputKey": i.get("fromOutputKey"), "uomId": i["uomId"], "qtyPerBatch": i["qtyPerBatch"]} for i in s["inputs"]], "outputs": [{"key": o["key"], "itemId": o.get("itemId"), "name": o["name"], "uomId": o["uomId"], "qtyPerBatch": o["qtyPerBatch"]} for o in s["outputs"]]} for s in tpl["stages"] ], "edges": [{"parentKey": e["parentKey"], "childKey": e["childKey"]} for e in tpl["edges"]], }, if_match=head.etag) chk.status("PUT template while a run is InProgress", locked, 409, "TEMPLATE_IN_USE") # Deactivating must stay allowed — it only blocks NEW runs (FR-MFG-01). chk.status("PATCH status while a run is InProgress is still allowed", c.patch(f"/production-templates/{tid}/status", {"status": "Inactive"}), 204) chk.status("run creation from an Inactive template", c.post("/production-runs", {"templateId": tid, "targetQty": 5, "warehouseId": warehouse}), 422, "TEMPLATE_INACTIVE") c.patch(f"/production-templates/{tid}/status", {"status": "Active"}) chk.check("activeRunCount incremented by the new run", next(t for t in c.get(f"/production-templates?q={TEMPLATE_CODE}").body["items"] if t["templateId"] == tid)["activeRunCount"], runs_before + 1) # ------------------------------------------------------------- quantities chk.section("7. Per-run quantity override (FR-MFG-08)") cut_input = cut["inputs"][0] edited = c.put(f"/production-runs/{rid}/stages/{cut['runStageId']}/quantities", {"inputs": [{"id": cut_input["runInputId"], "plannedQty": 420}], "outputs": []}) chk.status("PUT quantities on a Ready stage", edited, 200) if edited.status == 200: chk.check("plannedQty updated", float(edited.body["inputs"][0]["plannedQty"]), 420.0) chk.check("stage still Ready (no upstream inputs)", edited.body["status"], "Ready") chk.status("PUT quantities naming another stage's input", c.put(f"/production-runs/{rid}/stages/{cut['runStageId']}/quantities", {"inputs": [{"id": asm["inputs"][0]["runInputId"], "plannedQty": 9}], "outputs": []}), 422) # Raising a Waiting stage's upstream input must keep it Waiting, and the edit must be # rejected outright once a stage has started (covered in M4 once we can start one). asm_up = next(i for i in asm["inputs"] if i["source"] == "Upstream") bumped = c.put(f"/production-runs/{rid}/stages/{asm['runStageId']}/quantities", {"inputs": [{"id": asm_up["runInputId"], "plannedQty": 60}], "outputs": []}) chk.status("PUT quantities on a Waiting stage", bumped, 200) if bumped.status == 200: chk.check("stage stays Waiting (nothing delivered)", bumped.body["status"], "Waiting") chk.status("PUT quantities on a nonexistent stage", c.put(f"/production-runs/{rid}/stages/999999/quantities", {"inputs": [], "outputs": []}), 404) # ------------------------------------------------------------------ events chk.section("8. Event history records the edits") detail = c.get(f"/production-runs/{rid}") chk.status("GET /production-runs/{id}", detail, 200) events = detail.body["events"] # Two, not four: only the two successful edits are recorded. The 422 (wrong stage) and # the 404 both throw inside ExecuteInTransactionAsync, so their event write rolls back # with the rest of the change — history never shows an edit that did not happen. chk.check("only successful edits are logged", sum(1 for e in events if e["eventType"] == "QuantityEdit"), 2) chk.check("event payload captured", events[0]["payload"] is not None if events else False, True) chk.check("event carries an actor", events[0]["userId"] > 0 if events else False, True) print(f"\nleft run {rid} ({run['docNo']}) InProgress for the M4 smoke test") return chk.finish("M3") if __name__ == "__main__": sys.exit(main())