dodil/erp-brick-sales-order

DODIL ERP brick: sales order

tests/contract_checks.py

"""Every call this brick makes to a neighbour is checked against the CALLEE's signed spec (erp-method). A missing checkout is an
error, never a skip. The fakes (which replace whole ports and never reach a Seam) call `spec_check` with the exact request the real
adapter would send; the real Seam calls are checked by `checked_seams` (conftest)."""
import os
import pathlib
import re

from erp_platform import contract as _contract
from erp_platform.contract import Recorder

HERE = pathlib.Path(__file__).resolve().parent
METHOD_DIR = os.environ.get("ERP_METHOD_DIR") or str(HERE.parents[1] / "erp-method")
if not (pathlib.Path(METHOD_DIR) / "modules" / "shared.yaml").exists():
    raise RuntimeError(f"seam contract checks need an erp-method checkout: set ERP_METHOD_DIR (tried {METHOD_DIR})")

CONTRACT = Recorder()

# Routes this brick calls that the callee's signed spec does not list for sales-order (reported to the method as spec gaps). Only the
# "no seam / listed for" finding on these is let through; a body that breaks a LISTED route's shape still fails.
# v7 of ar, inventory and purchasing list every route this brick calls (seam-gap PR 2), so those exceptions are gone. Two are left, both on the
# CALLEE's own spec (reported to the method): product lists GET /products/{id}/convert for inventory and purchasing only, and business-partner
# lists GET /partners/{bp_id} for ar, payments, purchasing and tax, not for sales-order.
SPEC_GAPS = (("product", r"GET /products/[^/]+/convert"), ("business-partner", r"GET /partners/\d+"))


# CHECKER MISREADS, reported (not hidden): ar's POST /schedules request prose marks a field "(optional)" only on some, and the checker reads an
# unmarked field as required (a schedule's per-line copies). ar's POST /invoices/raise now has a request_schema, so its entry is gone.
OPTIONAL_IN_PROSE = {("ar", "POST /schedules"): {"quantity", "discount_pct", "unit_code", "product_id", "description", "tax_category", "in_advance"}}


_check = _contract.check


def _check_but_known_gaps(callee, method, path, body=None, **kw):
    bad = _check(callee, method, path, body, **kw)
    if any(callee == c and re.fullmatch(p, f"{method} {path}") for c, p in SPEC_GAPS):
        bad = [b for b in bad if "is listed for" not in b and "no seam" not in b]
    skip = OPTIONAL_IN_PROSE.get((callee, f"{method} {path}"), set())
    return [b for b in bad if not (b.strip().endswith(": required by the spec and missing") and b.split(":")[0].strip() in skip)]


_contract.check = _check_but_known_gaps          # the real Seam calls (checked_seams) and the fakes below are filtered the same way


def spec_check(callee, method, path, body=None):
    bad = _contract.check(callee, method, path, body, caller="sales-order", method_dir=METHOD_DIR)
    if bad:
        msg = f"sales-order -> {callee} {method} {path} breaks the callee's spec:\n  " + "\n  ".join(bad)
        CONTRACT.violations.append(msg)
        raise AssertionError(msg)


def inbound_check(caller, method, path, body=None):
    """A neighbour that CALLS this brick (timesheets, SO-51) is checked on both sides: the caller's own signed outbound entry (the
    request it says it sends) and this brick's inbound entry naming it. A fake of that neighbour sends only what passes both."""
    bad = _contract.check("sales-order", method, path, body, caller=caller, method_dir=METHOD_DIR)
    spec, _shared = _contract.load_callee(caller, METHOD_DIR)
    mine = [s for s in spec.get("seams") or [] if s.get("with") == "sales-order" and s.get("direction") == "outbound"
            and s.get("route", "").startswith(method)]
    if not mine:
        bad.append(f"{caller}: its signed spec lists no outbound {method} seam with sales-order")
    for s in mine:
        for f in s.get("request_schema") or []:
            if f.get("required") and f.get("in") != "path" and f["field"] not in (body or {}):
                bad.append(f"{caller}'s own spec requires {f['field']} and the call omits it")
            if f.get("values") and f["field"] in (body or {}) and body[f["field"]] not in f["values"]:
                bad.append(f"{f['field']}={body[f['field']]!r} is not one of {f['values']} in {caller}'s spec")
    if bad:
        msg = f"{caller} -> sales-order {method} {path} breaks the signed specs:\n  " + "\n  ".join(bad)
        CONTRACT.violations.append(msg)
        raise AssertionError(msg)