"""M2 smoke test — production template CRUD + graph validation (docs/30 §D.1, FR-MFG-01..07). Touches no stock: templates only. Run with the API and AuthHex up: python Backend/smoke/m2_templates.py Shape under test is a diamond, which exercises multiple entries converging on one terminal *and* a stage with two parents: Cut ─┐ ├─▶ Assemble (terminal, real item) Prep ─┘ """ from __future__ import annotations import sys from smoke_common import bootstrap TEMPLATE_CODE = "SMOKE-PT-M2" def pick_fixtures(c, chk): """Grab two active items and a UOM to build a realistic template from.""" items = c.get("/items?pageSize=5&status=Active").body["items"] if len(items) < 2: sys.exit("FATAL: need at least 2 active items in the database.") uoms = c.get("/uoms?pageSize=5").body["items"] if not uoms: sys.exit("FATAL: need at least 1 UOM in the database.") return items[0]["itemId"], items[1]["itemId"], uoms[0]["uomId"] def diamond(raw_item, finished_item, uom): """A valid graph: two entry stages feeding one terminal stage.""" return { "code": TEMPLATE_CODE, "name": "Smoke chair line", "description": "Created by m2_templates.py", "stages": [ { "key": "tmp-cut", "name": "Cut frame", "roleLabel": "Carpentry", "estimatedMinutes": 60, "posX": 80, "posY": 120, "fieldDefs": [{"key": "moisture_ok", "label": "Moisture check", "type": "Checkbox", "required": True}], "inputs": [{"source": "Stock", "itemId": raw_item, "qtyPerBatch": 8}], "outputs": [{"key": "tmp-frame", "name": "Frame set", "uomId": uom, "qtyPerBatch": 1}], }, { "key": "tmp-prep", "name": "Prep cushions", "roleLabel": "Upholstery", "estimatedMinutes": 30, "posX": 80, "posY": 320, "fieldDefs": [], "inputs": [{"source": "Stock", "itemId": raw_item, "qtyPerBatch": 2}], "outputs": [{"key": "tmp-cushion", "name": "Cushion set", "uomId": uom, "qtyPerBatch": 1}], }, { "key": "tmp-asm", "name": "Assemble & QA", "roleLabel": "QA", "estimatedMinutes": 45, "posX": 560, "posY": 200, "fieldDefs": [], "inputs": [ {"source": "Upstream", "fromOutputKey": "tmp-frame", "qtyPerBatch": 1}, {"source": "Upstream", "fromOutputKey": "tmp-cushion", "qtyPerBatch": 1}, ], # Terminal output must name the finished item (FR-MFG-05), and takes its unit # from that item — sending a uomId as well is rejected. "outputs": [{"key": "tmp-chair", "name": "Chair", "itemId": finished_item, "qtyPerBatch": 1}], }, ], "edges": [ {"parentKey": "tmp-cut", "childKey": "tmp-asm"}, {"parentKey": "tmp-prep", "childKey": "tmp-asm"}, ], # Canvas-only decoration. No graph semantics at all — the validator never sees these, # which is exactly what the "annotations do not affect the graph" assertion checks. "annotations": [ {"kind": "box", "posX": 40, "posY": 60, "width": 400, "height": 320, "label": "Sub-assembly", "rotation": None}, {"kind": "line", "posX": 480, "posY": 40, "width": 220, "height": 4, "label": "Phase 2", "rotation": 90}, ], } def cleanup(c): """ Find any template this script left behind, and clear the edit-lock if later scripts started runs against it. Without this the script is single-use: FR-MFG-06 refuses a PUT while any run of the template is InProgress, so a second execution would fail its very first assertion with 409 TEMPLATE_IN_USE. Cancelling those runs is safe — they belong to the smoke suite. """ existing = c.get(f"/production-templates?q={TEMPLATE_CODE}").body["items"] template = next((t for t in existing if t["code"] == TEMPLATE_CODE), None) if template is None: return None tid = template["templateId"] if template["activeRunCount"] > 0: reason = next((r["reasonCodeId"] for r in c.get("/reason-codes?context=Production&pageSize=50").body["items"] if r["code"] == "PRD-CANCEL"), None) blocking = [r for r in c.get(f"/production-runs?templateId={tid}&status=InProgress&pageSize=200").body["items"]] for run in blocking: c.post(f"/production-runs/{run['runId']}/cancel", {"reasonCodeId": reason, "note": "cancelled by m2_templates.py to clear the edit-lock"}) print(f"cancelled {len(blocking)} in-progress run(s) to release the template edit-lock") return tid def main(): c, chk, args = bootstrap(__doc__) print(f"API {args.api} · AuthHex {args.auth}") raw_item, finished_item, uom = pick_fixtures(c, chk) print(f"fixtures: rawItem={raw_item} finishedItem={finished_item} uom={uom}") stale = cleanup(c) if stale: print(f"note: reusing/overwriting existing template {stale} ({TEMPLATE_CODE})") payload = diamond(raw_item, finished_item, uom) # ---------------------------------------------------------------- create chk.section("1. Create a valid diamond graph") if stale: head = c.get(f"/production-templates/{stale}") created = c.put(f"/production-templates/{stale}", payload, if_match=head.etag) chk.status("PUT existing template", created, 200) else: created = c.post("/production-templates", payload) chk.status("POST /production-templates", created, 201) if created.status not in (200, 201): return chk.finish("M2") tid = created.body["templateId"] chk.check("ETag header present", created.etag is not None, True) chk.check("3 stages persisted", len(created.body["stages"]), 3) chk.check("2 edges persisted", len(created.body["edges"]), 2) # ------------------------------------------------------------------- get chk.section("2. GET round-trips ids, keys, positions and fieldDefs") got = c.get(f"/production-templates/{tid}") chk.status("GET /production-templates/{id}", got, 200) g = got.body stages = {s["name"]: s for s in g["stages"]} chk.check("stage keys equal their ids", all(s["key"] == str(s["stageId"]) for s in g["stages"]), True) cut = stages["Cut frame"] chk.check("posX survived the round trip", float(cut["posX"]), 80.0) chk.check("fieldDefs jsonb survived", cut["fieldDefs"][0]["key"], "moisture_ok") chk.check("fieldDef type survived as enum name", cut["fieldDefs"][0]["type"], "Checkbox") asm = stages["Assemble & QA"] chk.check("terminal has 2 upstream inputs", sum(1 for i in asm["inputs"] if i["source"] == "Upstream"), 2) chk.check("upstream inputs resolved fromOutputId", all(i["fromOutputId"] for i in asm["inputs"] if i["source"] == "Upstream"), True) chk.check("upstream inputs expose fromOutputKey", all(i["fromOutputKey"] for i in asm["inputs"] if i["source"] == "Upstream"), True) chk.check("terminal output carries the finished item", asm["outputs"][0]["itemId"], finished_item) # The builder reads activeRunCount straight off the graph to decide whether to disable # itself; without it on this response it would need a second request to the list endpoint. chk.check("graph exposes activeRunCount", g["activeRunCount"], 0) # Canvas annotations. Not in docs/30 Part C — added so a save can't silently discard the # boxes and dividers the builder already draws. chk.section("2b. Canvas annotations round-trip") anns = {a["kind"]: a for a in g["annotations"]} chk.check("both annotations persisted", len(g["annotations"]), 2) chk.check("box label survived", anns.get("box", {}).get("label"), "Sub-assembly") chk.check("box geometry survived", (float(anns["box"]["posX"]), float(anns["box"]["width"])), (40.0, 400.0)) chk.check("line rotation survived", float(anns.get("line", {}).get("rotation") or 0), 90.0) chk.check("annotations are not stages", len(g["stages"]), 3) bad_kind = dict(rebuild_from_get(g)) bad_kind["annotations"] = [{"kind": "circle", "posX": 0, "posY": 0, "width": 10, "height": 10, "label": None, "rotation": None}] chk.status("unknown annotation kind rejected", c.put(f"/production-templates/{tid}", bad_kind, if_match=got.etag), 422) # The keys GET hands back must be directly reusable as a PUT payload. chk.section("3. Idempotent re-save using the server's own keys") echo = rebuild_from_get(g) resaved = c.put(f"/production-templates/{tid}", echo, if_match=got.etag) chk.status("PUT echoing server keys", resaved, 200) if resaved.status == 200: chk.check("stage ids preserved (diffed, not recreated)", sorted(s["stageId"] for s in resaved.body["stages"]), sorted(s["stageId"] for s in g["stages"])) chk.check("edge count still 2", len(resaved.body["edges"]), 2) chk.check("annotations survived the echo re-save", len(resaved.body["annotations"]), 2) etag = resaved.etag # Wholesale replacement cuts both ways: a client that forgets to echo annotations back # wipes them. Pinned explicitly because that is a silent data loss, not an error. stripped = rebuild_from_get(g) stripped["annotations"] = [] cleared = c.put(f"/production-templates/{tid}", stripped, if_match=etag) chk.status("PUT omitting annotations", cleared, 200) if cleared.status == 200: chk.check("omitted annotations are cleared", len(cleared.body["annotations"]), 0) restored = c.put(f"/production-templates/{tid}", rebuild_from_get(g), if_match=cleared.etag) chk.status("PUT restoring annotations", restored, 200) etag = restored.etag if restored.status == 200 else cleared.etag else: etag = got.etag # ------------------------------------------------------- graph rejections chk.section("4. Graph validation rejections (FR-MFG-02, FR-MFG-04, FR-MFG-05)") cycle = rebuild_from_get(g) cycle["edges"].append({"parentKey": key_of(g, "Assemble & QA"), "childKey": key_of(g, "Cut frame")}) chk.status("cycle", c.put(f"/production-templates/{tid}", cycle, if_match=etag), 422, "GRAPH_CYCLE") two_term = rebuild_from_get(g) two_term["edges"] = [e for e in two_term["edges"] if e["parentKey"] != key_of(g, "Prep cushions")] chk.status("two terminals", c.put(f"/production-templates/{tid}", two_term, if_match=etag), 422, "GRAPH_TERMINAL_COUNT") # An isolated stage has no outbound edge, so it is *also* a second terminal and the # cheaper terminal-count check catches it first. That is the more useful error anyway — # it names both offending stages. GRAPH_DISCONNECTED is unreachable in a valid-so-far # DAG (see the note in ProductionGraphValidator) and is kept only as defence in depth. orphan = rebuild_from_get(g) orphan["stages"].append({ "key": "tmp-orphan", "name": "Orphan stage", "estimatedMinutes": 5, "posX": 900, "posY": 600, "fieldDefs": [], "inputs": [], "outputs": [{"key": "tmp-orphan-out", "name": "Nothing", "uomId": uom, "qtyPerBatch": 1}], }) chk.status("isolated stage (reported as a second terminal)", c.put(f"/production-templates/{tid}", orphan, if_match=etag), 422, "GRAPH_TERMINAL_COUNT") # Grandparent reference: Cut -> Mid -> Asm, with Asm drawing from Cut's output. grandparent = rebuild_from_get(g) cut_key = key_of(g, "Cut frame") asm_key = key_of(g, "Assemble & QA") cut_output_key = output_key_of(g, "Cut frame", "Frame set") grandparent["stages"].append({ "key": "tmp-mid", "name": "Middle", "estimatedMinutes": 5, "posX": 320, "posY": 120, "fieldDefs": [], "inputs": [{"source": "Upstream", "fromOutputKey": cut_output_key, "qtyPerBatch": 1}], "outputs": [{"key": "tmp-mid-out", "name": "Mid part", "uomId": uom, "qtyPerBatch": 1}], }) grandparent["edges"] = [e for e in grandparent["edges"] if e["parentKey"] != cut_key] grandparent["edges"] += [{"parentKey": cut_key, "childKey": "tmp-mid"}, {"parentKey": "tmp-mid", "childKey": asm_key}] # Assemble still reads Cut's output, but Cut is now a grandparent -> invalid. chk.status("upstream input from a grandparent", c.put(f"/production-templates/{tid}", grandparent, if_match=etag), 422, "GRAPH_INPUT_SOURCE_INVALID") no_item = rebuild_from_get(g) for s in no_item["stages"]: if s["name"] == "Assemble & QA": s["outputs"][0]["itemId"] = None chk.status("terminal output without an item", c.put(f"/production-templates/{tid}", no_item, if_match=etag), 422, "TERMINAL_OUTPUT_ITEM_REQUIRED") wip_item = rebuild_from_get(g) for s in wip_item["stages"]: if s["name"] == "Cut frame": s["outputs"][0]["itemId"] = finished_item chk.status("intermediate output claiming an item", c.put(f"/production-templates/{tid}", wip_item, if_match=etag), 422) bad_item = rebuild_from_get(g) for s in bad_item["stages"]: if s["name"] == "Cut frame": s["inputs"][0]["itemId"] = 999_999 chk.status("stock input naming a nonexistent item", c.put(f"/production-templates/{tid}", bad_item, if_match=etag), 422) chk.section("5. Concurrency + status") chk.status("PUT with no If-Match", c.request("PUT", f"/production-templates/{tid}", rebuild_from_get(g)), 428, "PRECONDITION_REQUIRED") # Malformed vs stale are different failures. "AQAAAA==" is a well-formed 4-byte token # (xmin = 1) that no live row will ever carry, so it reaches the service's version # comparison and yields 412 rather than the 428 a garbage token would. chk.status("PUT with a malformed If-Match", c.put(f"/production-templates/{tid}", rebuild_from_get(g), if_match='"not-base64"'), 428, "PRECONDITION_REQUIRED") chk.status("PUT with a stale If-Match", c.put(f"/production-templates/{tid}", rebuild_from_get(g), if_match='"AQAAAA=="'), 412, "CONCURRENCY_CONFLICT") chk.status("PATCH status -> Inactive", c.patch(f"/production-templates/{tid}/status", {"status": "Inactive"}), 204) listed = c.get(f"/production-templates?q={TEMPLATE_CODE}") row = next((t for t in listed.body["items"] if t["templateId"] == tid), None) chk.check("list reports Inactive", row and row["status"], "Inactive") chk.check("list reports stageCount 3", row and row["stageCount"], 3) chk.check("list reports activeRunCount 0", row and row["activeRunCount"], 0) # Leave it Active so the M3 run-creation smoke can start runs from it. c.patch(f"/production-templates/{tid}/status", {"status": "Active"}) print(f"\nleft template {tid} ({TEMPLATE_CODE}) Active for the M3 smoke test") return chk.finish("M2") # --- helpers: turn a GET response back into a save payload ------------------- def rebuild_from_get(g: dict) -> dict: """Echo a fetched graph back as a save payload, reusing the server's keys.""" return { "code": g["code"], "name": g["name"], "description": g.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"), "qtyUnit": i["qtyUnit"], "qtyPerBatch": i["qtyPerBatch"]} for i in s["inputs"] ], # uomId round-trips as null on an item-bearing output and as the WIP label # otherwise, so echoing it back verbatim is correct either way. "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 g["stages"] ], "edges": [{"parentKey": e["parentKey"], "childKey": e["childKey"]} for e in g["edges"]], # Echoed back deliberately: annotations are replaced wholesale, so omitting them here # would make every re-save silently clear the canvas layout notes. "annotations": g["annotations"], } def key_of(g: dict, stage_name: str) -> str: return next(s["key"] for s in g["stages"] if s["name"] == stage_name) def output_key_of(g: dict, stage_name: str, output_name: str) -> str: stage = next(s for s in g["stages"] if s["name"] == stage_name) return next(o["key"] for o in stage["outputs"] if o["name"] == output_name) if __name__ == "__main__": sys.exit(main())