"""contract.py — check a brick's outgoing seam calls against the CALLEE's signed spec.
Why it exists: the costliest cross-brick bugs were calls our tests accepted and the real callee would
refuse — a journal event the ledger does not take, an integer where the ledger's id is text. Each brick's
tests talk to a hand-written fake that follows the CALLER's guess, so the guess and the fake agree and
nothing fails until a real run. This module reads the callee's `modules/<id>/module.yaml` `seams:` entry
for the route (the `request:` text) and `modules/shared.yaml` (key types), and says what the spec says
the call must look like.
What it checks, and only what the spec text states: the route + method is listed for that callee; a
field not marked optional is present (`a | b` = one of them); a type (a shared key's VARCHAR/BIGINT, or
the spec's own "BIGINT as a string", "string", "YYYY-MM-DD", "ISO 4217"); a value inside the
enumeration (`event (approve | pay | reverse)`, `event: issue | void`) or equal to a quoted literal
(`source_module: "ar"`); and `journal_id` equal to the `derived_id(...)` the spec names. Lines
(`lines: [{...}]`) are checked per element. A field the spec does not mention is NOT flagged (spec
prose is not guaranteed exhaustive).
Where the seam entry carries a structured `request_schema:` (erp-method pattern.md section 13) the schema is
checked EXACTLY instead of the prose: required, type (a shared key's declared type or a basic type), `values`,
array `items`, `derived`, `in: path | query`, and a field the schema does not list is flagged. Query fields are
(an array travels comma-separated) checked on GET/DELETE (the params); on a body verb only body fields are seen. The prose heuristic is the fallback
for entries without a schema.
Opt in with ONE line in a brick's tests/conftest.py (needs ERP_METHOD_DIR, or method_dir=, pointing at an
erp-method checkout):
from erp_platform.contract import checked_seams
checked_seams = checked_seams(caller="fixed-assets", aliases={"fa": "fixed-assets"}) # autouse
Every `Seam` call then goes through `check`; a violation raises ContractViolation at the call and is
re-asserted at teardown (so a brick that swallows exceptions still fails). A brick whose tests replace
the whole port with a fake (never reaching Seam) wraps that fake instead: `guard(monkeypatch, fake,
"post_journal", callee="gl", method="POST", path="/journal-entry/journals", caller="purchasing")`.
"""
import os
import re
from datetime import date
from functools import lru_cache
from pathlib import Path
from typing import Any, Optional
import yaml
from erp_platform.seam import Seam, derived_id
try: # the C loader is ~10x faster on a 2,400-line module.yaml
_Loader = yaml.CSafeLoader
except AttributeError: # pragma: no cover
_Loader = yaml.SafeLoader
class ContractViolation(AssertionError):
"""A seam call the callee's spec says the callee would refuse."""
# ── loading ───────────────────────────────────────────────────────────────────────────────
def _modules_dir(method_dir: Optional[str]) -> Path:
base = Path(method_dir or os.environ.get("ERP_METHOD_DIR", ""))
if not str(base):
raise ContractViolation("no erp-method directory: pass method_dir= or set ERP_METHOD_DIR")
return base / "modules" if (base / "modules").is_dir() else base
@lru_cache(maxsize=None)
def _load(path: str) -> dict:
with open(path, encoding="utf-8") as f:
return yaml.load(f, Loader=_Loader) or {}
def load_callee(callee: str, method_dir: Optional[str] = None) -> tuple:
"""(module.yaml dict, shared keys dict) for a module id, or from a module.yaml path."""
if callee.endswith((".yaml", ".yml")):
p = Path(callee)
mods = p.parent.parent
else:
mods = _modules_dir(method_dir)
p = mods / callee / "module.yaml"
shared = mods / "shared.yaml"
return _load(str(p)), (_load(str(shared)).get("keys") or {} if shared.exists() else {})
# ── parsing the spec's request text ───────────────────────────────────────────────────────
def _split(text: str, sep: str = ",") -> list:
"""Split on `sep` at bracket depth 0, outside double quotes."""
out, depth, cur, quoted = [], 0, [], False
for ch in text:
if ch == '"':
quoted = not quoted
elif not quoted:
if ch in "{[(":
depth += 1
elif ch in "}])":
depth -= 1
elif ch == sep and depth == 0:
out.append("".join(cur))
cur = []
continue
cur.append(ch)
out.append("".join(cur))
return [s.strip() for s in out if s.strip()]
def _matching(text: str, start: int) -> int:
depth = 0
for i in range(start, len(text)):
depth += text[i] in "{[("
depth -= text[i] in "}])"
if depth == 0:
return i
return len(text) - 1
_NAMES = re.compile(r"^([a-z][a-z0-9_]*(?:\s*(?:\||\bOR\b)\s*[a-z][a-z0-9_]*)*)")
_WORD = re.compile(r"^[a-z][a-z_]*$")
class Field:
def __init__(self, item: str, shared: dict, nested: bool = False):
self.text = item
m = _NAMES.match(item)
self.names = re.split(r"\s*(?:\||\bOR\b)\s*", m.group(1)) if m else []
rest = item[m.end():].strip() if m else ""
# The spec marks "optional" and never "required". Top level: unmarked = required. Inside a list element
# (a journal line) fields differ per line kind, so only a one-of group (`debit | credit`) is required.
self.optional = bool(re.search(r"optional|absent|omitted|\bv3\b", item)) or (nested and len(self.names) < 2)
self.children: list = [] # element variants of `lines: [{...} | {...}]`
self.enum: Optional[set] = None
self.literal: Optional[str] = None
self.kind: Optional[str] = None
self.derived = re.findall(r"derived_id\(([^)]*)\)", item) if "journal_id" in self.names else []
value = rest[1:].strip() if rest.startswith(":") else rest
if value.startswith("[{"): # a list of objects: parse each variant
inner = value[1:_matching(value, 0)]
self.children = [parse_object(v, shared, True) for v in _split(inner, "|")]
elif value.startswith("{"):
self.children = [parse_object(value, shared, True)]
# enumeration: `name: a | b | c` or `name (a | b | c)`
cand = value if rest.startswith(":") else (value[1:_matching(value, 0)] if value.startswith("(") else "")
cand = re.sub(r"\([^)]*\)", "", cand)
alts = [a.strip() for a in cand.split("|")]
if len(alts) > 1 and all(_WORD.match(a) and not a.endswith("_id") for a in alts):
self.enum = set(alts)
lit = re.match(r'^"([^"<>:]+)"\s*(?:\(|$)', value) if rest.startswith(":") else None
if lit:
self.literal = lit.group(1)
# type: the spec's own wording first, else the shared key's declared type
name = self.names[0] if len(self.names) == 1 else ""
head = re.sub(r"\[.*\]|\{.*\}", "", item) # a hint inside a nested list describes the elements, not this field
if re.search(r"BIGINT as a string|BIGINT as string|str of (?:a|the) BIGINT", head):
self.kind = "bigint_str"
elif re.search(r"\(string\b|:\s*string\b", head):
self.kind = "str"
elif "YYYY-MM-DD" in head:
self.kind = "date"
elif "ISO 4217" in head and self.enum is None:
self.kind = "ccy"
elif name in shared and shared[name].get("type") in ("VARCHAR", "BIGINT"):
self.kind = "str" if shared[name]["type"] == "VARCHAR" else "bigint"
def parse_object(text: str, shared: dict, nested: bool = False) -> list:
text = text.strip()
if text.startswith("{") and text.endswith("}"):
text = text[1:-1]
return [f for f in (Field(i, shared, nested) for i in _split(text)) if f.names]
def parse_request(request: str, shared: dict) -> list:
"""The fields a request text declares: a `{...}` body, or `query a=.., b=..` parameters."""
req = (request or "").strip()
if req.startswith("{"):
return parse_object(req, shared)
m = re.search(r"\bquery\s+(.*)$", req)
if not m:
return []
# a `<...>` placeholder describes ONE value (`invoice_ids=<comma-separated, at most 200>`): its words are not fields
q = re.sub(r"<[^<>]*>", "<value>", m.group(1))
items = [re.sub(r"^([a-z0-9_]+)=", r"\1: ", i) for i in _split(q)]
return [f for f in (Field(i, shared) for i in items) if f.names]
# ── checking ──────────────────────────────────────────────────────────────────────────────
def _kind_ok(kind: str, v: Any) -> bool:
if kind == "str":
return isinstance(v, str)
if kind == "bigint": # the spec does not say whether a shared BIGINT key travels as number or string: accept both
return (isinstance(v, int) and not isinstance(v, bool)) or (isinstance(v, str) and v.lstrip("-").isdigit())
if kind == "bigint_str":
return isinstance(v, str) and v.lstrip("-").isdigit()
if kind == "date":
return isinstance(v, date) or (isinstance(v, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", v) is not None)
if kind == "ccy":
return isinstance(v, str) and re.fullmatch(r"[A-Z]{3}", v) is not None
return True
_KIND_WORDS = {"str": "text", "bigint": "an integer (or its digits)", "bigint_str": "a BIGINT as a string",
"date": "a YYYY-MM-DD date", "ccy": "an ISO 4217 code"}
def _check_fields(fields: list, obj: Any, where: str) -> list:
if not isinstance(obj, dict):
return [f"{where}: expected an object, got {type(obj).__name__}"]
bad = []
for f in fields:
label = f"{where}{' | '.join(f.names)}"
present = [n for n in f.names if obj.get(n) is not None]
if not present:
if not f.optional:
bad.append(f"{label}: required by the spec and missing")
continue
for n in present:
v = obj[n]
if f.kind and len(f.names) == 1 and not _kind_ok(f.kind, v):
bad.append(f"{where}{n}: spec says {_KIND_WORDS[f.kind]}, got {type(v).__name__} {v!r}")
if f.enum is not None and n == f.names[-1] and v not in f.enum:
bad.append(f"{where}{n}: {v!r} is not accepted; the spec lists {' | '.join(sorted(f.enum))}")
if f.literal is not None and len(f.names) == 1 and v != f.literal:
bad.append(f"{where}{n}: the spec fixes this to {f.literal!r}, got {v!r}")
if f.children and isinstance(v, list):
for i, el in enumerate(v):
tries = [_check_fields(c, el, f"{where}{n}[{i}].") for c in f.children]
if all(tries):
bad.extend(min(tries, key=len))
if "journal_id" in obj and obj["journal_id"] is not None:
f = next((x for x in fields if x.names == ["journal_id"]), None)
for args in sorted((f.derived if f else []), key=lambda a: -len(a.split(","))):
names = [a.strip() for a in args.split(",")]
if all(obj.get(n) is not None for n in names):
want = str(derived_id(*[obj[n] for n in names]))
if str(obj["journal_id"]) != want:
bad.append(f"{where}journal_id: the spec says derived_id({', '.join(names)}) = {want}, got {obj['journal_id']!r}")
break
return bad
# ── the structured schema (`request_schema:`), checked exactly ────────────────────────────
_PRIMITIVES = {"string", "integer", "decimal", "date", "currency", "boolean", "object", "array"}
def _type_ok(tp: str, v: Any, shared: dict, query: bool = False) -> Optional[bool]:
"""True/False for a value against a schema type; None when the type is unknown to both vocabularies."""
if tp in shared:
st = str(shared[tp].get("type", "")).upper()
if st.startswith("BIGINT"):
return _kind_ok("bigint", v)
if st.startswith("VARCHAR") or st.startswith("TEXT"):
return isinstance(v, str)
return True
if tp == "string":
return isinstance(v, str)
if tp == "integer":
return (isinstance(v, int) and not isinstance(v, bool)) or (query and isinstance(v, str) and v.lstrip("-").isdigit())
if tp == "decimal":
if isinstance(v, bool):
return False
if isinstance(v, (int, float)) or type(v).__name__ == "Decimal":
return True
try:
float(v)
return isinstance(v, str)
except (TypeError, ValueError):
return False
if tp == "date":
return _kind_ok("date", v)
if tp == "currency":
return _kind_ok("ccy", v)
if tp == "boolean":
return isinstance(v, bool) or (query and v in ("true", "false"))
if tp == "object":
return isinstance(v, dict)
if tp == "array":
return isinstance(v, list)
return None
def _check_schema(fields: list, obj: Any, shared: dict, where: str = "", *, loc: Optional[str] = None,
query: bool = False) -> list:
"""Exact check of `obj` against a list of {field, type, required, values, items, derived} specs.
`loc` selects which fields apply: None = body fields, "query" = query params; unknown keys are flagged."""
if not isinstance(obj, dict):
return [f"{where}: expected an object, got {type(obj).__name__}"]
mine = [f for f in fields if f.get("in") == loc] if loc else [f for f in fields if not f.get("in")]
bad, known = [], {str(f["field"]) for f in mine}
for f in mine:
n, tp = str(f["field"]), f.get("type")
v = obj.get(n)
if v is None:
if f.get("required"):
bad.append(f"{where}{n}: required by the spec and missing")
continue
if query and tp == "array" and isinstance(v, str): # a query array travels comma-separated: `ids=1,2,3`
v = [x for x in v.split(",") if x != ""]
ok = _type_ok(tp, v, shared, query) if tp else True
if ok is None:
bad.append(f"{where}{n}: the spec's own type {tp!r} is neither a shared key nor a basic type")
elif not ok:
bad.append(f"{where}{n}: spec says {tp}, got {type(v).__name__} {v!r}")
if f.get("values") is not None and v not in f["values"] and str(v) not in [str(x) for x in f["values"]]:
bad.append(f"{where}{n}: {v!r} is not accepted; the spec lists {' | '.join(map(str, f['values']))}")
if f.get("items") is not None and isinstance(v, list):
items = f["items"]
for i, el in enumerate(v):
if items and isinstance(items[0], dict):
bad.extend(_check_schema(items, el, shared, f"{where}{n}[{i}].", query=query))
elif items and not any(_type_ok(t, el, shared, query) for t in items):
bad.append(f"{where}{n}[{i}]: {el!r} is none of the spec's item types {items}")
if f.get("derived"):
m = re.match(r"derived_id\(([^)]*)\)", str(f["derived"]))
names = [a.strip() for a in m.group(1).split(",")] if m else []
if names and all(obj.get(a) is not None for a in names):
want = str(derived_id(*[obj[a] for a in names]))
if str(v) != want:
bad.append(f"{where}{n}: the spec says derived_id({', '.join(names)}) = {want}, got {v!r}")
bad.extend(f"{where}{k}: not in the spec's request_schema" for k in sorted(set(obj) - known))
return bad
def _schema_for(entry: dict, method: str, path: str, body: Optional[dict], shared: dict) -> list:
fields = entry["request_schema"]
bad = []
tmpl = entry["route"].split(None, 1)[1].split("?", 1)[0]
names = re.findall(r"\{([^}]+)\}", tmpl)
if names:
m = re.match("^" + re.sub(r"\\\{[^}]+\\\}", lambda x: "([^/]+)", re.escape(tmpl)) + "$", path)
got = dict(zip(names, m.groups())) if m else {}
for f in fields:
if f.get("in") == "path":
v = got.get(str(f["field"]))
if v is not None and _type_ok(f.get("type"), v, shared, True) is False:
bad.append(f"path {f['field']}: spec says {f.get('type')}, got {v!r}")
if v is not None and f.get("values") is not None and v not in [str(x) for x in f["values"]]:
bad.append(f"path {f['field']}: {v!r} is not accepted; the spec lists {' | '.join(map(str, f['values']))}")
if method in ("POST", "PUT", "PATCH"):
bad += _check_schema(fields, body or {}, shared)
else:
bad += _check_schema(fields, body or {}, shared, loc="query", query=True)
return bad
def _route_regex(route: str) -> re.Pattern:
path = route.split("?", 1)[0]
return re.compile("^" + re.sub(r"\\\{[^}]+\\\}", "[^/]+", re.escape(path)) + "$")
def check(callee: str, method: str, path: str, body: Optional[dict] = None, *,
caller: Optional[str] = None, method_dir: Optional[str] = None) -> list:
"""Violations of the callee's spec by this call; [] when it conforms.
`callee`: a module id (resolved under method_dir / $ERP_METHOD_DIR) or a module.yaml path. `body` is the JSON
body (POST/PUT) or the query params (GET/DELETE). `caller` narrows to the seam entries that name that module
(`with:`), which matters where one route serves many callers with different shapes (gl's journals).
"""
spec, shared = load_callee(callee, method_dir)
cid = spec.get("id") or (Path(callee).parent.name if callee.endswith(".yaml") else callee)
method, path = method.upper(), path.split("?", 1)[0]
entries = [s for s in spec.get("seams") or [] if s.get("owner") in (None, cid)
and re.match(r"^[A-Z]+\s+/", s.get("route", ""))]
hits = [s for s in entries if s["route"].split()[0] == method and _route_regex(s["route"].split(None, 1)[1]).match(path)]
if not hits:
listed = sorted({s["route"] for s in entries})
return [f"{cid}: no seam {method} {path} in the spec (listed: {'; '.join(listed)})"]
named = [s for s in hits if caller and caller in _callers(s)]
if caller and not named:
return [f"{cid}: {method} {path} is listed for {sorted({s.get('with', '?') for s in hits})}, not for {caller}"]
results = [_schema_for(s, method, path, body, shared) if s.get("request_schema")
else _check_fields(parse_request(s.get("request", ""), shared), body or {}, "") for s in (named or hits)]
return [] if any(not r for r in results) else min(results, key=len)
def _callers(seam: dict) -> list:
"""The modules a seam entry names in `with:` — one id, a list, or "expenses, pos" (0.9.8)."""
w = seam.get("with")
items = w if isinstance(w, list) else str(w or "").split(",")
return [str(x).strip() for x in items if str(x).strip()]
# ── plugging into tests ───────────────────────────────────────────────────────────────────
class Recorder:
def __init__(self):
self.violations: list = []
def verify(self, callee, method, path, body, caller, method_dir):
bad = check(callee, method, path, body, caller=caller, method_dir=method_dir)
if bad:
msg = f"{caller or '?'} -> {callee} {method} {path} breaks the callee's spec:\n " + "\n ".join(bad)
self.violations.append(msg)
raise ContractViolation(msg)
def checked_seams(caller: str, aliases: Optional[dict] = None, method_dir: Optional[str] = None,
skip: tuple = ()) -> Any:
"""An autouse pytest fixture: every `Seam` call is checked against the callee's spec.
`aliases` maps a Seam's name to its module id where they differ ({"fa": "fixed-assets", "inv": "inventory"}).
A seam whose module has no module.yaml (or is listed in `skip`) is not checked. Assign it to a module-level
name in conftest.py so pytest collects it.
"""
import pytest
me = caller
@pytest.fixture(autouse=True)
def _checked_seams(monkeypatch):
rec = Recorder()
original = Seam._call
def call(self, method, path, actor, caller=None, **kw):
mod = (aliases or {}).get(self.name, self.name)
if mod not in skip and self.installed and _has_spec(mod, method_dir):
rec.verify(mod, method, path, kw.get("json") if method in ("POST", "PUT", "PATCH") else kw.get("params"),
me, method_dir)
return original(self, method, path, actor, caller, **kw)
monkeypatch.setattr(Seam, "_call", call)
yield rec
assert not rec.violations, "\n".join(rec.violations)
return _checked_seams
def _has_spec(mod: str, method_dir: Optional[str]) -> bool:
return (_modules_dir(method_dir) / mod / "module.yaml").exists()
def guard(monkeypatch, fake: Any, attr: str, *, callee: str, method: str, path: str, caller: str,
method_dir: Optional[str] = None, recorder: Optional[Recorder] = None) -> Recorder:
"""Wrap a port fake's method `attr(body)` so its first argument is checked as the body of `method path`."""
rec, inner = recorder or Recorder(), getattr(fake, attr)
def wrapped(body, *a, **kw):
rec.verify(callee, method, path, body, caller, method_dir)
return inner(body, *a, **kw)
monkeypatch.setattr(fake, attr, wrapped)
return rec