ERP-core/Backend/smoke/smoke_common.py

"""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, 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, "qty": q, "unitCost": cost, "discountPct": 0, "vatPct": 0}
            for (i, 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}")