New_REP_SYSTEM/ERP-core
8
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Refactor production and stock tests to remove UOM dependency

Browse Source 
- Updated production.spec.ts, stock-adjustments.spec.ts, and stock-transfers.spec.ts to eliminate UOM references in API seeder and test cases.
- Adjusted ApiSeeder methods to remove UOM parameters from stock receiving and production template creation.
- Revised documentation to reflect changes in UOM handling, emphasizing that stock is counted in base UOM only.
- Introduced new enums for MeasureUnit and StageQtyUnit to clarify content size and stage input quantities.
- Implemented ItemContent service to validate and normalize content sizes.
- Updated smoke tests to validate production stage inputs expressed in content units, ensuring correct consumption calculations.
- Modified frontend UOM label handling to reflect the removal of per-line UOMs in document lines.
This commit is contained in:
ImanThiyanga
2026-08-11 11:32:40 +05:30
parent d37824cecc
commit 15ddac178c
108 changed files with 1221 additions and 987 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Backend/smoke/m2_templates.py
+11 -8
View File
@@ -44,25 +44,26 @@ def diamond(raw_item, finished_item, uom):
"estimatedMinutes": 60, "posX": 80, "posY": 120,
"fieldDefs": [{"key": "moisture_ok", "label": "Moisture check",
"type": "Checkbox", "required": True}],
"inputs": [{"source": "Stock", "itemId": raw_item, "uomId": uom, "qtyPerBatch": 8}],
"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, "uomId": uom, "qtyPerBatch": 2}],
"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", "uomId": uom, "qtyPerBatch": 1},
{"source": "Upstream", "fromOutputKey": "tmp-cushion", "uomId": uom, "qtyPerBatch": 1},
{"source": "Upstream", "fromOutputKey": "tmp-frame", "qtyPerBatch": 1},
{"source": "Upstream", "fromOutputKey": "tmp-cushion", "qtyPerBatch": 1},
],
# Terminal output must name the finished item (FR-MFG-05).
# 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,
"uomId": uom, "qtyPerBatch": 1}],
"qtyPerBatch": 1}],
},
],
"edges": [
@@ -243,7 +244,7 @@ def main():
grandparent["stages"].append({
"key": "tmp-mid", "name": "Middle", "estimatedMinutes": 5, "posX": 320, "posY": 120,
"fieldDefs": [],
"inputs": [{"source": "Upstream", "fromOutputKey": cut_output_key, "uomId": uom, "qtyPerBatch": 1}],
"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]
@@ -320,9 +321,11 @@ def rebuild_from_get(g: dict) -> dict:
"inputs": [
{"source": i["source"], "itemId": i.get("itemId"),
"fromOutputKey": i.get("fromOutputKey"),
"uomId": i["uomId"], "qtyPerBatch": i["qtyPerBatch"]}
"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"]}
Backend/smoke/m3_runs.py
+1 -1
View File
@@ -147,7 +147,7 @@ def main():
"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"],
"fromOutputKey": i.get("fromOutputKey"), "qtyUnit": i["qtyUnit"],
"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"]]}
Backend/smoke/m4_stage_actions.py
+9 -9
View File
@@ -65,16 +65,16 @@ def ledger_rows(c, run_id, source):
return rows
def seed_stock(c, wh, raw, pack, uom):
def seed_stock(c, wh, raw, pack):
"""
Seed on-hand at explicit unit costs. Two raw layers at different costs mean the FIFO
consumption at start has to weight them, so `consumedValue` is a real number the
assertions can check rather than the 0.00 a positive adjustment would produce.
"""
seed_costed_stock(c, wh, [
(raw, uom, SEED_RAW * 0.4, RAW_COST_1),
(raw, uom, SEED_RAW * 0.6, RAW_COST_2),
(pack, uom, SEED_PACK, PACK_COST),
(raw, SEED_RAW * 0.4, RAW_COST_1),
(raw, SEED_RAW * 0.6, RAW_COST_2),
(pack, SEED_PACK, PACK_COST),
])
@@ -90,7 +90,7 @@ def build_template(c, raw, pack, finished, uom):
"estimatedMinutes": 60, "posX": 80, "posY": 100,
"fieldDefs": [{"key": "moisture_ok", "label": "Moisture check",
"type": "Checkbox", "required": True}],
"inputs": [{"source": "Stock", "itemId": raw, "uomId": uom, "qtyPerBatch": 8}],
"inputs": [{"source": "Stock", "itemId": raw, "qtyPerBatch": 8}],
"outputs": [{"key": "tmp-frame", "name": "Frame", "uomId": uom, "qtyPerBatch": 1}],
},
{
@@ -99,11 +99,11 @@ def build_template(c, raw, pack, finished, uom):
"key": "tmp-asm", "name": "Assemble", "roleLabel": "QA",
"estimatedMinutes": 45, "posX": 520, "posY": 100, "fieldDefs": [],
"inputs": [
{"source": "Upstream", "fromOutputKey": "tmp-frame", "uomId": uom, "qtyPerBatch": 1},
{"source": "Stock", "itemId": pack, "uomId": uom, "qtyPerBatch": 2},
{"source": "Upstream", "fromOutputKey": "tmp-frame", "qtyPerBatch": 1},
{"source": "Stock", "itemId": pack, "qtyPerBatch": 2},
],
"outputs": [{"key": "tmp-chair", "name": "Chair", "itemId": finished,
"uomId": uom, "qtyPerBatch": 1}],
"qtyPerBatch": 1}],
},
],
"edges": [{"parentKey": "tmp-cut", "childKey": "tmp-asm"}],
@@ -141,7 +141,7 @@ def main():
drained = drain_stock(c, wh)
if drained:
print(f"drained {len(drained)} leftover item(s) from a previous execution")
seed_stock(c, wh, raw, pack, uom)
seed_stock(c, wh, raw, pack)
raw_before = on_hand(c, raw, wh)
pack_before = on_hand(c, pack, wh)
print(f"warehouse={wh} raw={raw}(on-hand {raw_before}) pack={pack}(on-hand {pack_before}) finished={finished}")
Backend/smoke/m4b_uom_conversion.py
-170
View File
@@ -1,170 +0,0 @@
"""M4b smoke test — UOM conversion on production stock inputs.
This covers the single highest-risk correctness gap in the manufacturing phase.
`IFifoCostingService.ConsumeAsync` works exclusively in an item's BASE UOM, while
`STAGE_INPUT.uom_id` is a free FK — docs/30 never mentions conversion at all. Without the
shared `IUomConverter` (extracted from `GrnService.ToBaseAsync`), a stage input declared in
"box of 12" would consume 1 base unit instead of 12 and silently mis-cost the whole run.
The dev database has no `uom_conversions` rows at all, so the non-base path was previously
unexercised by any data. This script creates a real conversion and proves:
* a stage input in a non-base UOM consumes qtyPerBatch x scaleFactor x factor base units
* the ledger records the BASE quantity, not the declared one
* an input in a UOM with no conversion defined is refused with 422 rather than mis-consumed
python Backend/smoke/m4b_uom_conversion.py
"""
from __future__ import annotations
import sys
from smoke_common import bootstrap
WAREHOUSE_CODE = "SMOKE-PRD"
TEMPLATE_CODE = "SMOKE-PT-M4B"
FACTOR = 12 # 1 case = 12 base units
QTY_PER_BATCH = 3 # cases per batch
TARGET_QTY = 10 # -> scale 10 -> 30 cases -> 360 base units
SEED = 5000
def main():
c, chk, args = bootstrap(__doc__)
print(f"API {args.api}")
# --- fixtures ---------------------------------------------------------
wh = next((w["warehouseId"] for w in c.get(f"/warehouses?q={WAREHOUSE_CODE}&pageSize=50").body["items"]
if w["code"] == WAREHOUSE_CODE), None)
if wh is None:
sys.exit("FATAL: run m4_stage_actions.py first (it creates the SMOKE-PRD warehouse).")
items = c.get("/items?pageSize=5&status=Active").body["items"]
raw, finished = items[0], items[1]
base_uom = raw["baseUomId"]
uoms = c.get("/uoms?pageSize=50").body["items"]
case_uom = next((u["uomId"] for u in uoms if u["uomId"] != base_uom), None)
if case_uom is None:
sys.exit("FATAL: need at least 2 UOMs to test conversion.")
print(f"item={raw['itemId']} baseUom={base_uom} caseUom={case_uom} factor={FACTOR}")
# --- define the conversion -------------------------------------------
chk.section("1. Define a non-base UOM conversion for the item")
conv = c.put(f"/items/{raw['itemId']}/uom-conversions",
{"conversions": [{"fromUom": case_uom, "toUom": base_uom, "factor": FACTOR}]})
chk.status("PUT /items/{id}/uom-conversions", conv, 200)
if conv.status != 200:
return chk.finish("M4b")
chk.check("conversion stored", any(float(x["factor"]) == FACTOR for x in conv.body["conversions"]), True)
reason = c.get("/reason-codes?context=Adjustment&pageSize=5").body["items"][0]["reasonCodeId"]
c.post("/stock-adjustments", {"warehouseId": wh, "reasonCodeId": reason,
"lines": [{"itemId": raw["itemId"], "qtyDelta": SEED}]})
before = float(c.get(f"/stock/on-hand?itemId={raw['itemId']}&warehouseId={wh}").body["onHand"])
print(f"on-hand before: {before}")
# --- template whose stock input is declared in CASES ------------------
chk.section("2. A stage input declared in the non-base UOM")
payload = {
"code": TEMPLATE_CODE, "name": "Smoke M4b conversion line",
"stages": [{
"key": "tmp-only", "name": "Pack", "estimatedMinutes": 10,
"posX": 0, "posY": 0, "fieldDefs": [],
# Declared in cases, not base units.
"inputs": [{"source": "Stock", "itemId": raw["itemId"],
"uomId": case_uom, "qtyPerBatch": QTY_PER_BATCH}],
"outputs": [{"key": "tmp-out", "name": "Packed", "itemId": finished["itemId"],
"uomId": base_uom, "qtyPerBatch": 1}],
}],
"edges": [],
}
existing = next((t for t in c.get(f"/production-templates?q={TEMPLATE_CODE}").body["items"]
if t["code"] == TEMPLATE_CODE), None)
if existing:
head = c.get(f"/production-templates/{existing['templateId']}")
res = c.put(f"/production-templates/{existing['templateId']}", payload, if_match=head.etag)
tid = existing["templateId"] if res.status in (200, 409) else None
chk.check("template ready", tid is not None, True)
else:
res = c.post("/production-templates", payload)
chk.status("create single-stage template", res, 201)
tid = res.body["templateId"] if res.status == 201 else None
if tid is None:
return chk.finish("M4b")
# A lone stage is both the entry and the terminal — worth asserting explicitly.
run = c.post("/production-runs", {"templateId": tid, "targetQty": TARGET_QTY, "warehouseId": wh})
chk.status("create the run", run, 201)
if run.status != 201:
return chk.finish("M4b")
stage = run.body["stages"][0]
chk.check("single stage is both entry and terminal",
(stage["isEntry"], stage["isTerminal"]), (True, True))
chk.check("single stage starts Ready", stage["status"], "Ready")
chk.check("plannedQty stays in the DECLARED uom (3 x 10 cases)",
float(stage["inputs"][0]["plannedQty"]), float(QTY_PER_BATCH * TARGET_QTY))
# --- the actual conversion assertion ---------------------------------
chk.section("3. Consumption converts cases to base units")
expected_base = QTY_PER_BATCH * TARGET_QTY * FACTOR # 3 x 10 x 12 = 360
started = c.post(f"/production-runs/{run.body['runId']}/stages/{stage['runStageId']}/start")
chk.status("start the stage", started, 200)
if started.status != 200:
return chk.finish("M4b")
con = started.body["consumed"][0]
chk.check(f"consumed {expected_base} BASE units, not {QTY_PER_BATCH * TARGET_QTY}",
float(con["qty"]), float(expected_base))
chk.check("on-hand fell by the base quantity",
float(c.get(f"/stock/on-hand?itemId={raw['itemId']}&warehouseId={wh}").body["onHand"]),
before - expected_base)
rows = c.get(f"/stock/ledger?sourceDocType=PRDI&sourceDocId={run.body['runId']}&pageSize=50").body["items"]
chk.check("one PRDI row", len(rows), 1)
if rows:
chk.check("ledger qtyBase is the converted quantity", float(rows[0]["qtyBase"]), float(expected_base))
detail = c.get(f"/production-runs/{run.body['runId']}").body
chk.check("consumedQty stored in base units",
float(detail["stages"][0]["inputs"][0]["consumedQty"]), float(expected_base))
# --- missing conversion is refused, not silently mis-consumed --------
chk.section("4. An undefined conversion is refused (422), never assumed 1:1")
third_uom = next((u["uomId"] for u in c.get("/uoms?pageSize=50").body["items"]
if u["uomId"] not in (base_uom, case_uom)), None)
if third_uom is None:
chk.check("skipped: need a third UOM", True, True)
else:
bad = dict(payload)
bad["code"] = TEMPLATE_CODE + "-BAD"
bad["stages"] = [dict(payload["stages"][0])]
bad["stages"][0] = {**payload["stages"][0],
"inputs": [{"source": "Stock", "itemId": raw["itemId"],
"uomId": third_uom, "qtyPerBatch": 1}]}
made = c.post("/production-templates", bad)
if made.status != 201:
head = c.get(f"/production-templates?q={TEMPLATE_CODE}-BAD")
tid2 = next((t["templateId"] for t in head.body["items"]
if t["code"] == TEMPLATE_CODE + "-BAD"), None)
else:
tid2 = made.body["templateId"]
if tid2:
run2 = c.post("/production-runs", {"templateId": tid2, "targetQty": 1, "warehouseId": wh})
if run2.status == 201:
s2 = run2.body["stages"][0]["runStageId"]
chk.status("start a stage whose input UOM has no conversion",
c.post(f"/production-runs/{run2.body['runId']}/stages/{s2}/start"), 422)
else:
chk.check("could create the second run", run2.status, 201)
return chk.finish("M4b")
if __name__ == "__main__":
sys.exit(main())
Backend/smoke/m4c_content_units.py
+182
View File
@@ -0,0 +1,182 @@
"""M4c smoke test — production stage inputs expressed in content units (FR-MFG-04).
Replaces m4b_uom_conversion.py, which tested the per-item UOM conversion table that no longer
exists. Stock is now always a count of the item's base UOM (a bottle, a packet), and the only
place another unit appears is a production formula: a stage input may be written as an amount
of the item's *content* (ml or g), which the server divides by the item's content size to get
the pack count FIFO actually consumes.
What this pins down, none of which any other script covers:
* 2000 ml against a 500 ml bottle consumes exactly 4.0000 bottles, and the ledger agrees;
* 300 ml against the same item consumes 0.6000 — **fractional packs are legal**, which is
the whole reason the quantity columns are (18,4);
* a Content input on an item with no content size is refused at template save (422), not
silently treated as packs at stage start.
Self-contained: creates its own item, warehouse, template and runs.
python Backend/smoke/m4c_content_units.py
"""
from __future__ import annotations
import sys
from smoke_common import bootstrap, drain_stock, seed_costed_stock
WAREHOUSE_CODE = "SMOKE-CONTENT"
TEMPLATE_CODE = "SMOKE-PT-M4C"
SKU = "SMOKE-BOTTLE-500ML"
PLAIN_SKU = "SMOKE-NO-CONTENT"
CONTENT_QTY = 500 # ml held by one bottle
SEED_BOTTLES = 100
UNIT_COST = 3.0
def ensure_warehouse(c):
for w in c.get(f"/warehouses?q={WAREHOUSE_CODE}&pageSize=50").body["items"]:
if w["code"] == WAREHOUSE_CODE:
return w["warehouseId"]
created = c.post("/warehouses", {"code": WAREHOUSE_CODE, "name": "Content-unit smoke warehouse"})
if created.status != 201:
sys.exit(f"FATAL: could not create the smoke warehouse: {created.status} {created.body}")
return created.body["warehouseId"]
def ensure_item(c, sku, name, content_qty, content_unit):
"""Find-or-create; content fields are only sent when the item is meant to have them."""
for i in c.get(f"/items?q={sku}&pageSize=50").body["items"]:
if i["sku"] == sku:
return i
category = c.get("/categories?pageSize=1").body["items"]
uoms = c.get("/uoms?pageSize=1").body["items"]
if not category or not uoms:
sys.exit("FATAL: need at least one category and one UOM seeded.")
body = {
"sku": sku, "name": name,
"categoryId": category[0]["categoryId"],
"baseUomId": uoms[0]["uomId"],
"stockNature": "Stocked", "trackingMode": "None",
}
if content_qty is not None:
body["contentQty"] = content_qty
body["contentUnit"] = content_unit
created = c.post("/items", body)
if created.status != 201:
sys.exit(f"FATAL: could not create item {sku}: {created.status} {created.body}")
return created.body
def save_template(c, raw_item, finished_item, qty_per_batch, qty_unit):
payload = {
"code": TEMPLATE_CODE, "name": "Content-unit smoke line",
"stages": [{
"key": "tmp-mix", "name": "Mix", "estimatedMinutes": 5, "posX": 0, "posY": 0,
"fieldDefs": [],
"inputs": [{"source": "Stock", "itemId": raw_item,
"qtyUnit": qty_unit, "qtyPerBatch": qty_per_batch}],
"outputs": [{"key": "tmp-out", "name": "Mixed", "itemId": finished_item,
"qtyPerBatch": 1}],
}],
"edges": [],
}
existing = next((t for t in c.get(f"/production-templates?q={TEMPLATE_CODE}").body["items"]
if t["code"] == TEMPLATE_CODE), None)
if existing:
head = c.get(f"/production-templates/{existing['templateId']}")
return c.put(f"/production-templates/{existing['templateId']}", payload, if_match=head.etag)
return c.post("/production-templates", payload)
def saved_ok(chk, label, response):
"""Template save is find-or-create, so a POST gives 201 and a PUT gives 200."""
ok = chk.check(f"{label} -> 200/201", response.status in (200, 201), True)
if not ok:
print(f" server said: {response.body}")
return ok
def consume_once(c, chk, tid, wh, target_qty, label, expected_packs):
"""
Create a run, start its only stage, and assert what FIFO actually took.
`expected_packs` is compared against the RAW formula quantity resolved to packs, which is
only valid while the run scale factor is 1. Run creation computes
`ratio = targetQty / terminalOutput.qtyPerBatch` and pre-scales every planned quantity, so
this holds because `save_template` pins the terminal output to `qtyPerBatch: 1` and every
caller here passes `target_qty=1`. Change either and these numbers move by that ratio.
"""
created = c.post("/production-runs", {"templateId": tid, "targetQty": target_qty, "warehouseId": wh})
if not chk.status(f"{label}: create the run", created, 201):
return
run = created.body
stage_id = run["stages"][0]["runStageId"]
started = c.post(f"/production-runs/{run['runId']}/stages/{stage_id}/start",
idempotency_key=f"m4c-{run['runId']}")
if not chk.status(f"{label}: POST .../start", started, 200):
return
consumed = started.body["consumed"]
if not chk.check(f"{label}: exactly one input consumed", len(consumed), 1):
return
chk.check(f"{label}: consumed {expected_packs} base units",
round(float(consumed[0]["qty"]), 4), expected_packs)
rows = c.get(f"/stock/ledger?sourceDocType=PRDI&sourceDocId={run['runId']}&pageSize=50").body["items"]
chk.check(f"{label}: one PRDI ledger row", len(rows), 1)
if rows:
chk.check(f"{label}: ledger qtyBase matches the consumption",
round(abs(float(rows[0]["qtyBase"])), 4), expected_packs)
def main():
c, chk, args = bootstrap(__doc__)
print(f"API {args.api}")
bottle = ensure_item(c, SKU, "Smoke syrup 500ml bottle", CONTENT_QTY, "Ml")
plain = ensure_item(c, PLAIN_SKU, "Smoke item with no content", None, None)
finished = c.get("/items?pageSize=5&status=Active").body["items"][0]
print(f"bottle={bottle['itemId']} contentBaseQty={bottle.get('contentBaseQty')} "
f"plain={plain['itemId']} finished={finished['itemId']}")
chk.section("0. Content size normalises to a base unit on the item")
chk.check("contentBaseQty is the entered ml", float(bottle["contentBaseQty"]), float(CONTENT_QTY))
chk.check("contentBaseUnit is Ml", bottle["contentBaseUnit"], "Ml")
wh = ensure_warehouse(c)
drain_stock(c, wh)
seed_costed_stock(c, wh, [(bottle["itemId"], SEED_BOTTLES, UNIT_COST)])
# ------------------------------------------------- whole packs out of content units
chk.section("1. A content quantity resolves to whole packs (2000 ml / 500 ml = 4)")
saved = save_template(c, bottle["itemId"], finished["itemId"], 2000, "Content")
if saved_ok(chk, "save the Content template", saved):
consume_once(c, chk, saved.body["templateId"], wh, 1, "2000 ml", 4.0)
# --------------------------------------------------------------- fractional packs
chk.section("2. A content quantity below one pack consumes a FRACTION of one (300 ml = 0.6)")
saved = save_template(c, bottle["itemId"], finished["itemId"], 300, "Content")
if saved_ok(chk, "save the fractional Content template", saved):
consume_once(c, chk, saved.body["templateId"], wh, 1, "300 ml", 0.6)
# ------------------------------------------------------------------ the guard rail
chk.section("3. Content units are refused on an item that has no content size")
refused = save_template(c, plain["itemId"], finished["itemId"], 100, "Content")
chk.status("Content input on a contentless item", refused, 422)
chk.section("4. The same item still works when the formula is written in packs")
saved = save_template(c, bottle["itemId"], finished["itemId"], 3, "Pack")
if saved_ok(chk, "save the Pack template", saved):
consume_once(c, chk, saved.body["templateId"], wh, 1, "3 bottles", 3.0)
return chk.finish("M4c")
if __name__ == "__main__":
sys.exit(main())
Backend/smoke/m5_receipt.py
+6 -6
View File
@@ -157,18 +157,18 @@ def main():
if i["itemId"] == raw)
finished_b = state["finishedItemId"]
# A fresh single-stage template: base-UOM input so no conversion muddies the arithmetic,
# qtyPerBatch 3.5 so the consumed quantity is NOT a multiple of the target (which is what
# A fresh single-stage template: a pack-counted input, so no content division muddies
# the arithmetic; qtyPerBatch 3.5 so consumed is NOT a multiple of the target (which is what
# forces pool / goodQty to repeat).
payload_b = {
"code": BIG_TEMPLATE_CODE, "name": "Smoke M5 rounding line",
"stages": [{
"key": "tmp-b", "name": "Mix", "estimatedMinutes": 5, "posX": 0, "posY": 0,
"fieldDefs": [],
"inputs": [{"source": "Stock", "itemId": raw, "uomId": raw_uom,
"inputs": [{"source": "Stock", "itemId": raw,
"qtyPerBatch": BIG_QTY_PER_BATCH}],
"outputs": [{"key": "tmp-bo", "name": "Mixed", "itemId": finished_b,
"uomId": raw_uom, "qtyPerBatch": 1}],
"qtyPerBatch": 1}],
}],
"edges": [],
}
@@ -190,7 +190,7 @@ def main():
# and the pool would not match the figures this section reasons about — the assertions
# would still "pass" while testing something else entirely.
drain_stock(c, wh)
seed_costed_stock(c, wh, [(raw, raw_uom, 5000, BIG_UNIT_COST)])
seed_costed_stock(c, wh, [(raw, 5000, BIG_UNIT_COST)])
big = c.post("/production-runs", {"templateId": tid, "targetQty": BIG_TARGET, "warehouseId": wh})
if big.status != 201:
@@ -243,7 +243,7 @@ def main():
"key": "tmp-one", "name": "Make", "estimatedMinutes": 1, "posX": 0, "posY": 0,
"fieldDefs": [], "inputs": [],
"outputs": [{"key": "tmp-o", "name": "Tracked", "itemId": tracked["itemId"],
"uomId": tracked["baseUomId"], "qtyPerBatch": 1}],
"qtyPerBatch": 1}],
}],
"edges": [],
}
Backend/smoke/m6_m7_leftover_rework_cancel.py
+4 -4
View File
@@ -54,13 +54,13 @@ def ensure_template(c, raw, finished, uom):
"stages": [
{"key": "tmp-cut", "name": "Cut", "estimatedMinutes": 10, "posX": 0, "posY": 0,
"fieldDefs": [],
"inputs": [{"source": "Stock", "itemId": raw, "uomId": uom, "qtyPerBatch": RAW_QPB}],
"inputs": [{"source": "Stock", "itemId": raw, "qtyPerBatch": RAW_QPB}],
"outputs": [{"key": "tmp-f", "name": "Frame", "uomId": uom, "qtyPerBatch": 1}]},
{"key": "tmp-asm", "name": "Assemble", "estimatedMinutes": 10, "posX": 400, "posY": 0,
"fieldDefs": [],
"inputs": [{"source": "Upstream", "fromOutputKey": "tmp-f", "uomId": uom, "qtyPerBatch": 1}],
"inputs": [{"source": "Upstream", "fromOutputKey": "tmp-f", "qtyPerBatch": 1}],
"outputs": [{"key": "tmp-c", "name": "Chair", "itemId": finished,
"uomId": uom, "qtyPerBatch": 1}]},
"qtyPerBatch": 1}]},
],
"edges": [{"parentKey": "tmp-cut", "childKey": "tmp-asm"}],
}
@@ -101,7 +101,7 @@ def main():
uom = items[0]["baseUomId"]
drain_stock(c, wh)
seed_costed_stock(c, wh, [(raw, uom, SEED, UNIT_COST)])
seed_costed_stock(c, wh, [(raw, SEED, UNIT_COST)])
tid = ensure_template(c, raw, finished, uom)
consumed_units = RAW_QPB * TARGET # 50
consumed_value = consumed_units * UNIT_COST # 150.00
Backend/smoke/run_all.py
+1 -1
View File
@@ -22,7 +22,7 @@ SCRIPTS = [
("M2 templates + graph validation", "m2_templates.py"),
("M3 run creation / board / quantities", "m3_runs.py"),
("M4 stage start / complete / approve / transfer", "m4_stage_actions.py"),
("M4b UOM conversion on stock inputs", "m4b_uom_conversion.py"),
("M4c content-unit stage inputs + fractional packs", "m4c_content_units.py"),
("M5 terminal receipt + cost pool", "m5_receipt.py"),
("M6+M7 leftover / rework / cancel", "m6_m7_leftover_rework_cancel.py"),
]
Backend/smoke/smoke_common.py
+4 -3
View File
@@ -240,15 +240,16 @@ def seed_costed_stock(c, warehouse_id: int, lines, vendor_id: int | None = None)
"""
Create on-hand at explicit unit costs via a direct GRN + confirm.
`lines` is an iterable of (item_id, uom_id, qty, unit_cost).
`lines` is an iterable of (item_id, qty, unit_cost). Quantities are counts of the item's
base UOM — document lines carry no unit of their own.
"""
vendor_id = vendor_id or ensure_vendor(c)
grn = c.post("/grns", {
"vendorId": vendor_id,
"warehouseId": warehouse_id,
"lines": [
{"itemId": i, "uomId": u, "qty": q, "unitCost": cost, "discountPct": 0, "vatPct": 0}
for (i, u, q, cost) in lines
{"itemId": i, "qty": q, "unitCost": cost, "discountPct": 0, "vatPct": 0}
for (i, q, cost) in lines
],
})
if grn.status != 201: