"""FIFO stock-consumption tests for the MariaDB port (R4.6 in the migration plan). FifoCostingService.ConsumeAsync had to be redesigned, not translated: MariaDB has no `FOR UPDATE OF ` (MySQL 8.0.1+ only), and InnoDB's locking clause otherwise applies to every table a statement reads — so a single-statement `... NOT EXISTS (grn_lines) ... FOR UPDATE` would also X-lock grn_lines and batches, deadlocking against GrnService. It is now a two-step: unlocked candidate SELECT, then a locking read scoped to stock_layers, then a re-validation under the locks. Usage: python m9_fifo.py [--api URL] [--auth URL] [--user EMAIL] [--password PW] """ from __future__ import annotations import concurrent.futures import json import urllib.error import urllib.request import uuid from smoke_common import bootstrap def _req(client, method: str, path: str, body=None, headers=None): url = f"{client.api.rstrip('/')}{path}" data = json.dumps(body).encode() if body is not None else None r = urllib.request.Request(url, data=data, method=method) r.add_header("Authorization", f"Bearer {client.token}") if data is not None: r.add_header("Content-Type", "application/json") for k, v in (headers or {}).items(): r.add_header(k, v) try: with urllib.request.urlopen(r) as resp: raw = resp.read() return resp.status, (json.loads(raw) if raw else None) except urllib.error.HTTPError as e: raw = e.read() try: return e.code, (json.loads(raw) if raw else None) except json.JSONDecodeError: return e.code, raw.decode(errors="replace") def main() -> int: client, check, _args = bootstrap("MariaDB FIFO consumption tests") sfx = uuid.uuid4().hex[:8] print("\nFixtures") _, branches = _req(client, "GET", "/branches?page=1&pageSize=1") branch_id = ((branches or {}).get("items") or [{}])[0].get("branchId") st, wh = _req(client, "POST", "/warehouses", {"code": f"F-W-{sfx}", "name": f"FIFO WH {sfx}", "branchId": branch_id}) check.check("warehouse created", st, 201) wh_id = (wh or {}).get("warehouseId") _, cat = _req(client, "POST", "/categories", {"name": f"FIFO Cat {sfx}"}) _, uom = _req(client, "POST", "/uoms", {"name": f"F-U-{sfx}"}) st, item = _req(client, "POST", "/items", { "sku": f"F-SKU-{sfx}", "name": f"FIFO Item {sfx}", "categoryId": (cat or {}).get("categoryId"), "baseUomId": (uom or {}).get("uomId"), "stockNature": "Stocked", "trackingMode": "None"}) check.check("item created", st, 201) item_id = (item or {}).get("itemId") st, vendor = _req(client, "POST", "/vendors", {"code": f"F-V-{sfx}", "name": f"FIFO Vendor {sfx}", "currency": "LKR", "branchId": branch_id}) check.check("vendor created", st, 201) vendor_id = (vendor or {}).get("vendorId") print("\nInbound: two FIFO layers at different costs") grn_ids = [] for qty, cost in ((10, 100), (10, 150)): st, grn = _req(client, "POST", "/grns", { "vendorId": vendor_id, "warehouseId": wh_id, "branchId": branch_id, "lines": [{"itemId": item_id, "qty": qty, "unitCost": cost}]}) if st != 201: print(f" GRN create failed {st}: {grn}") check.check(f"GRN qty={qty} @ {cost} created", st, 201) break grn_id = (grn or {}).get("grnId") grn_ids.append(grn_id) st2, _ = _req(client, "POST", f"/grns/{grn_id}/confirm", {}) check.check(f"GRN {grn_id} (qty {qty} @ {cost}) confirmed", st2 in (200, 204), True) st, onhand = _req(client, "GET", f"/stock/on-hand?itemId={item_id}&warehouseId={wh_id}") print(f" on-hand response: {onhand}") print("\nFIFO consume via a negative adjustment (exercises the two-step locking read)") _, reasons = _req(client, "GET", "/reason-codes?context=Adjustment") rlist = (reasons or {}).get("items") or reasons or [] var = next((r for r in rlist if r.get("code") == "VAR"), None) check.check("seeded reason code VAR present", var is not None, True) reason_id = (var or {}).get("reasonCodeId") # Consume 15 of 20: must take all 10 @100 then 5 @150 => cost 1750, oldest-first. st, adj = _req(client, "POST", "/stock-adjustments", { "warehouseId": wh_id, "branchId": branch_id, "reasonCodeId": reason_id, "lines": [{"itemId": item_id, "qtyDelta": -15}]}) print(f" adjustment -> {st}: {json.dumps(adj)[:400] if adj else None}") check.check("negative adjustment accepted (FIFO consume ran)", st, 201) st, val = _req(client, "GET", f"/stock/valuation?itemId={item_id}&warehouseId={wh_id}") print(f" valuation: {json.dumps(val)[:400] if val else None}") if isinstance(val, dict): check.check("5 units remain after consuming 15 of 20", val.get("totalQty"), 5) # the survivor is the newer, dearer layer check.check("remaining value is 5 @ 150 = 750", val.get("totalValue"), 750) print("\nOver-issue must be blocked, not allowed negative") st, body = _req(client, "POST", "/stock-adjustments", { "warehouseId": wh_id, "branchId": branch_id, "reasonCodeId": reason_id, "lines": [{"itemId": item_id, "qtyDelta": -999}]}) code = (body or {}).get("code") if isinstance(body, dict) else None print(f" over-issue -> {st} code={code}") check.check("over-issue rejected with 409", st, 409) check.check("error code is STOCK_NEGATIVE_BLOCKED", code, "STOCK_NEGATIVE_BLOCKED") print("\nConcurrent issue of the last 5 units: exactly one wins") def issue(_): s, b = _req(client, "POST", "/stock-adjustments", { "warehouseId": wh_id, "branchId": branch_id, "reasonCodeId": reason_id, "lines": [{"itemId": item_id, "qtyDelta": -5}]}) return s, ((b or {}).get("code") if isinstance(b, dict) else None) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: out = list(pool.map(issue, range(2))) print(f" results={out}") oks = [s for s, _ in out if s == 201] check.check("exactly one issue succeeded", len(oks), 1) check.check("the loser was blocked, not a 500", all(s in (201, 409) for s, _ in out), True) st, val = _req(client, "GET", f"/stock/valuation?itemId={item_id}&warehouseId={wh_id}") if isinstance(val, dict): check.check("stock never went negative", (val.get("totalQty") or 0) >= 0, True) print(f" final qty={val.get('totalQty')}") print("\n" + "=" * 62) print(f" PASSED {check.passed} FAILED {check.failed}") print("=" * 62) return 1 if check.failed else 0 if __name__ == "__main__": raise SystemExit(main())