feat: add production templates API and documentation for manufacturing phase 2
- Implemented CRUD operations for production templates, including listing, retrieving, creating, updating, and deactivating templates. - Introduced a new API contract for production runs, detailing the lifecycle from creation to completion, including handling of stock inputs and outputs. - Documented the architecture, requirements, entity model, and API contract for the manufacturing phase 2, ensuring clarity on the production process and its integration with existing systems.
This commit is contained in:
@@ -0,0 +1,259 @@
|
||||
"""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}")
|
||||
Reference in New Issue
Block a user