"""Shared harness for the manufacturing smoke tests (docs/30-BACKEND-PHASE2.md). The repo has no test project; verification is a live smoke test against local Postgres with a real AuthHex session, with the assertion count recorded in Backend/PROGRESS.md. These scripts make that repeatable instead of ad-hoc curl. Auth note: ERPCore validates AuthHex's RS256 tokens offline against a statically configured public key, so we log in to AuthHex *directly* and send the access token as a Bearer header. That deliberately bypasses ERPCore's own /auth/login proxy, which would otherwise need AuthHex:BaseUrl to match the port AuthHex actually listens on. Usage: python m2_templates.py [--api URL] [--auth URL] [--user EMAIL] [--password PW] Environment variables (ERP_SMOKE_API, ERP_SMOKE_AUTH, ERP_SMOKE_USER, ERP_SMOKE_PASSWORD) override the defaults; command-line flags override those. """ from __future__ import annotations import argparse import json import os import sys import urllib.error import urllib.request # Server messages and box-drawing output contain non-cp1252 characters, and the default # Windows console codepage would raise UnicodeEncodeError mid-report — losing exactly the # diagnostic text a failing assertion needs to show. for _stream in (sys.stdout, sys.stderr): try: _stream.reconfigure(encoding="utf-8", errors="replace") except (AttributeError, ValueError): pass DEFAULTS = { "api": "http://localhost:5224/api/v1", "auth": "http://localhost:5602", "user": "admin@gmail.com", "password": "Naveen@99", } def parse_args(description: str) -> argparse.Namespace: p = argparse.ArgumentParser(description=description) for key, default in DEFAULTS.items(): p.add_argument(f"--{key}", default=os.environ.get(f"ERP_SMOKE_{key.upper()}", default)) return p.parse_args() class Response: __slots__ = ("status", "body", "headers") def __init__(self, status: int, body, headers: dict): self.status = status self.body = body self.headers = headers @property def code(self): """The RFC 7807 domain error code, when the body carries one (docs/11 §1.8).""" return self.body.get("code") if isinstance(self.body, dict) else None @property def etag(self): return self.headers.get("ETag") def __repr__(self): return f"<{self.status} code={self.code}>" class Client: def __init__(self, api: str, token: str): self.api = api.rstrip("/") self.token = token def request(self, method: str, path: str, body=None, if_match: str | None = None, idempotency_key: str | None = None) -> Response: url = f"{self.api}{path}" data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Accept", "application/json") req.add_header("Authorization", f"Bearer {self.token}") if data is not None: req.add_header("Content-Type", "application/json") if if_match: req.add_header("If-Match", if_match) if idempotency_key: req.add_header("Idempotency-Key", idempotency_key) try: with urllib.request.urlopen(req) as r: raw = r.read() return Response(r.status, _decode(raw), dict(r.headers)) except urllib.error.HTTPError as e: raw = e.read() return Response(e.code, _decode(raw), dict(e.headers)) def get(self, path): return self.request("GET", path) def post(self, path, body=None, **kw): return self.request("POST", path, body, **kw) def put(self, path, body=None, **kw): return self.request("PUT", path, body, **kw) def patch(self, path, body=None, **kw): return self.request("PATCH", path, body, **kw) def _decode(raw: bytes): if not raw: return None try: return json.loads(raw) except json.JSONDecodeError: return raw.decode(errors="replace") def login(auth_url: str, identifier: str, password: str) -> str: """Obtain an AuthHex access token via its `{functionName, payload}` envelope.""" payload = { "functionName": "loginUser", "payload": {"identifier": identifier, "password": password, "deviceName": "erp-smoke"}, "reference": "", } req = urllib.request.Request( f"{auth_url.rstrip('/')}/api/user", data=json.dumps(payload).encode(), method="POST", headers={"Content-Type": "application/json"}, ) try: with urllib.request.urlopen(req) as r: envelope = json.loads(r.read()) except urllib.error.URLError as e: sys.exit(f"FATAL: cannot reach AuthHex at {auth_url} ({e}). Start ERP_Auth_Service first.") if not envelope.get("success") or not envelope.get("data"): sys.exit(f"FATAL: AuthHex login failed: {envelope.get('message')!r}") return envelope["data"]["accessToken"] class Checker: """Counts assertions so the pass total can be recorded in PROGRESS.md.""" def __init__(self): self.passed = 0 self.failed = 0 def check(self, label: str, actual, expected) -> bool: ok = actual == expected if ok: self.passed += 1 print(f" PASS {label}") else: self.failed += 1 print(f" FAIL {label}\n expected: {expected!r}\n actual: {actual!r}") return ok def status(self, label: str, response: Response, expected_status: int, expected_code: str | None = None): ok = self.check(f"{label} -> {expected_status}", response.status, expected_status) if expected_code is not None: ok = self.check(f"{label} -> code {expected_code}", response.code, expected_code) and ok if not ok and response.status >= 400: detail = response.body.get("detail") if isinstance(response.body, dict) else response.body print(f" server said: {detail}") return ok def section(self, title: str): print(f"\n--- {title} ---") def finish(self, name: str) -> int: total = self.passed + self.failed print(f"\n{'=' * 60}\n{name}: {self.passed}/{total} assertions passed" + (f" ({self.failed} FAILED)" if self.failed else " — ALL GREEN") + f"\n{'=' * 60}") return 1 if self.failed else 0 def bootstrap(description: str): """Standard entry point: parse args, log in, return (client, checker, args).""" args = parse_args(description) token = login(args.auth, args.user, args.password) return Client(args.api, token), Checker(), args # --- stock fixtures ---------------------------------------------------------- # # Seeding matters more than it looks. A positive stock ADJUSTMENT is the obvious way to # create on-hand, but StockMutator's inbound path costs it at *last cost*, which is 0.00 # when the item has no prior layers. Stock seeded that way makes every cost-pool assertion # pass trivially against zeros and proves nothing. A direct (no-PO) GRN lets us state the # unit cost explicitly, so consumption produces a real, checkable value. def drain_stock(c, warehouse_id: int) -> list: """ Zero out every item's on-hand in a warehouse via one negative adjustment. Needed because these scripts are re-runnable and FIFO is oldest-first: stock left behind by a previous execution is consumed *before* anything seeded now. If an earlier run left zero-cost layers (as an adjustment-based seed does), a later run's cost assertions would silently read 0.00 and pass against nothing. Draining first makes each execution start from a known-empty warehouse. """ rows = c.get(f"/stock/on-hand/list?warehouseId={warehouse_id}&pageSize=200").body["items"] lines = [{"itemId": r["itemId"], "qtyDelta": -float(r["onHand"])} for r in rows if float(r["onHand"]) > 0] if not lines: return [] reasons = c.get("/reason-codes?context=Adjustment&pageSize=5").body["items"] if not reasons: sys.exit("FATAL: no Adjustment reason codes seeded.") res = c.post("/stock-adjustments", { "warehouseId": warehouse_id, "reasonCodeId": reasons[0]["reasonCodeId"], "lines": lines, }) if res.status != 201: sys.exit(f"FATAL: could not drain the smoke warehouse: {res.status} {res.body}") return lines def ensure_vendor(c) -> int: existing = c.get("/vendors?pageSize=1").body["items"] if existing: return existing[0]["vendorId"] created = c.post("/vendors", {"code": "SMOKE-V", "name": "Smoke vendor"}) if created.status != 201: sys.exit(f"FATAL: could not create a vendor: {created.status} {created.body}") return created.body["vendorId"] def seed_costed_stock(c, warehouse_id: int, lines, vendor_id: int | None = 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). """ 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 ], }) if grn.status != 201: sys.exit(f"FATAL: could not create the seeding GRN: {grn.status} {grn.body}") confirmed = c.post(f"/grns/{grn.body['grnId']}/confirm") if confirmed.status != 200: sys.exit(f"FATAL: could not confirm the seeding GRN: {confirmed.status} {confirmed.body}")