"""contract.py: a call is checked against the callee's signed spec — one test per violation kind."""
import pytest
import yaml
from erp_platform import contract, seam
from erp_platform.contract import ContractViolation, check
SPEC = {"id": "ledger", "seams": [
{"with": "buyer", "owner": "ledger", "route": "POST /journals",
"request": '{journal_id (BIGINT as string, derived_id(source_document, event)), source_module: "buyer", '
'source_document: bill_id | payment_id, event (approve | pay | reverse), journal_date: YYYY-MM-DD, '
'ledger_id (optional; absent = every ledger), currency: ISO 4217, '
'lines: [{account_id | purpose (clearing | tax), debit | credit, memo (optional)}]}'},
{"with": "buyer", "owner": "ledger", "route": "GET /periods/{company}/open",
"request": "path company; query date=YYYY-MM-DD, kind=spot | average (optional), bp_id (optional)"},
{"with": "buyer", "owner": "buyer", "route": "GET /bills/open", "request": "query as_of=YYYY-MM-DD"},
]}
SHARED = {"keys": {"ledger_id": {"type": "VARCHAR"}, "bp_id": {"type": "BIGINT"}}}
@pytest.fixture
def mdir(tmp_path):
(tmp_path / "modules" / "ledger").mkdir(parents=True)
(tmp_path / "modules" / "ledger" / "module.yaml").write_text(yaml.safe_dump(SPEC))
(tmp_path / "modules" / "shared.yaml").write_text(yaml.safe_dump(SHARED))
return str(tmp_path)
def good(**over):
b = {"journal_id": str(seam.derived_id("B-1", "approve")), "source_module": "buyer", "source_document": "B-1",
"event": "approve", "journal_date": "2026-10-01", "currency": "GBP",
"lines": [{"account_id": "10", "debit": "5.00"}, {"purpose": "tax", "credit": "5.00"}]}
b.update(over)
return b
def run(body, mdir, method="POST", path="/journals", **kw):
return check("ledger", method, path, body, caller="buyer", method_dir=mdir, **kw)
def test_a_good_call_passes(mdir):
assert run(good(), mdir) == []
assert run(good(ledger_id="IFRS"), mdir) == []
def test_unknown_route_and_method(mdir):
assert "no seam POST /nope" in run(good(), mdir, path="/nope")[0]
assert "no seam GET /journals" in run({}, mdir, method="GET", path="/journals")[0]
assert "/bills/open" not in run({}, mdir, path="/nope")[0] # a route the callee only CALLS is not its own
def test_route_with_path_params_matches(mdir):
assert run({"date": "2026-10-01"}, mdir, method="GET", path="/periods/UK01/open") == []
def test_missing_required_field_and_one_of(mdir):
b = good()
del b["currency"]
assert any("currency: required" in v for v in run(b, mdir))
b = good(lines=[{"memo": "x"}])
assert any("account_id | purpose" in v for v in run(b, mdir)) and any("debit | credit" in v for v in run(b, mdir))
assert run({}, mdir, method="GET", path="/periods/UK01/open")[0].startswith("date: required")
def test_wrong_type_including_shared_key_type(mdir):
assert any("ledger_id" in v and "text" in v for v in run(good(ledger_id=1), mdir)) # shared: VARCHAR
assert any("bp_id" in v for v in run({"date": "2026-10-01", "bp_id": "seven"}, mdir, method="GET", path="/periods/U/open"))
assert any("journal_id" in v and "BIGINT as a string" in v for v in run(good(journal_id=5), mdir))
assert any("journal_date" in v for v in run(good(journal_date="1 Oct"), mdir))
assert any("currency" in v for v in run(good(currency="pounds"), mdir))
def test_value_outside_the_accepted_set_and_literal(mdir):
assert any("'clear' is not accepted" in v and "approve | pay | reverse" in v for v in run(good(event="clear"), mdir))
assert any("fixes this to 'buyer'" in v for v in run(good(source_module="other"), mdir))
assert any("lines[0].purpose" in v for v in run(good(lines=[{"purpose": "bogus", "debit": "1"}]), mdir))
assert any("kind" in v for v in run({"date": "2026-10-01", "kind": "bogus"}, mdir, method="GET", path="/periods/U/open"))
def test_derived_id_mismatch(mdir):
bad = run(good(journal_id="123"), mdir)
assert any("derived_id(source_document, event)" in v for v in bad)
assert run(good(event="pay", journal_id=str(seam.derived_id("B-1", "pay"))), mdir) == []
def test_caller_must_be_named_by_the_seam(mdir):
assert "not for stranger" in check("ledger", "POST", "/journals", good(), caller="stranger", method_dir=mdir)[0]
def test_checked_seams_fixture_fails_a_bad_call_loudly(mdir, monkeypatch):
rec = contract.Recorder()
with pytest.raises(ContractViolation, match="clear"):
rec.verify("ledger", "POST", "/journals", good(event="clear"), "buyer", mdir)
assert rec.violations
def test_seam_call_is_intercepted(mdir, monkeypatch):
"""The fixture patches Seam._call: a bad body raises before anything leaves the process."""
monkeypatch.setenv("LEDGER_BASE_URL", "http://127.0.0.1:1")
fx = contract.checked_seams("buyer", method_dir=mdir).__wrapped__ if hasattr(contract.checked_seams("buyer"), "__wrapped__") else None
gen = fx(monkeypatch) if fx else None
if gen is None:
pytest.skip("pytest wraps fixtures without __wrapped__")
rec = next(gen)
s = seam.Seam("ledger", "LEDGER_BASE_URL", roles=[], permissions=[])
with pytest.raises(ContractViolation):
s.post("/journals", good(event="clear"))
assert s.post("/journals", good()).reason.startswith("ledger unreachable") # a good call goes on to the network
with pytest.raises(AssertionError):
next(gen)
assert len(rec.violations) == 1
def test_guard_wraps_a_port_fake(mdir, monkeypatch):
class Fake:
def post_journal(self, body):
return "ok"
f = Fake()
contract.guard(monkeypatch, f, "post_journal", callee="ledger", method="POST", path="/journals", caller="buyer", method_dir=mdir)
assert f.post_journal(good()) == "ok"
with pytest.raises(ContractViolation):
f.post_journal(good(event="refund"))
# ── request_schema: exact ──────────────────────────────────────────────────────────────────
SCHEMA_SPEC = {"id": "ledger", "seams": [
{"with": "buyer", "owner": "ledger", "route": "POST /journals", "request": "prose that would not be read",
"request_schema": [
{"field": "journal_id", "type": "journal_id", "required": False, "derived": "derived_id(source_document, event)"},
{"field": "source_document", "type": "string", "required": True},
{"field": "event", "type": "string", "required": True, "values": ["issue", "void"]},
{"field": "company_code", "type": "company_code", "required": False},
{"field": "amount", "type": "decimal", "required": False},
{"field": "tags", "type": "array", "required": False, "items": ["string"]},
{"field": "lines", "type": "array", "required": True, "items": [
{"field": "account_id", "type": "account_id", "required": False},
{"field": "debit", "type": "decimal", "required": False}]}]},
{"with": "buyer", "owner": "ledger", "route": "GET /fx/{currency}", "request": "x",
"request_schema": [
{"field": "currency", "type": "currency", "required": True, "in": "path"},
{"field": "on", "type": "date", "required": True, "in": "query"},
{"field": "kind", "type": "string", "required": False, "in": "query", "values": ["spot", "average"]},
{"field": "ids", "type": "array", "required": False, "in": "query", "items": ["integer"]}]},
]}
SCHEMA_SHARED = {"keys": {"journal_id": {"type": "BIGINT"}, "company_code": {"type": "VARCHAR"},
"account_id": {"type": "VARCHAR"}}}
@pytest.fixture
def sdir(tmp_path):
(tmp_path / "modules" / "ledger").mkdir(parents=True)
(tmp_path / "modules" / "ledger" / "module.yaml").write_text(yaml.safe_dump(SCHEMA_SPEC))
(tmp_path / "modules" / "shared.yaml").write_text(yaml.safe_dump(SCHEMA_SHARED))
return str(tmp_path)
def sgood(**over):
b = {"source_document": "D-1", "event": "issue", "lines": [{"account_id": "10", "debit": "5.00"}]}
b.update(over)
return b
def srun(body, sdir, method="POST", path="/journals"):
return check("ledger", method, path, body, caller="buyer", method_dir=sdir)
def test_schema_good_call_passes(sdir):
assert srun(sgood(company_code="UK01", amount=5, tags=["a"],
journal_id=str(seam.derived_id("D-1", "issue"))), sdir) == []
def test_schema_required_type_values_items_and_unknown(sdir):
assert any("source_document" in v and "missing" in v for v in srun(sgood(source_document=None), sdir))
assert any("company_code" in v for v in srun(sgood(company_code=7), sdir)) # VARCHAR shared key sent as int
assert any("'void'" not in v and "event" in v for v in srun(sgood(event="reverse"), sdir))
assert any("amount" in v for v in srun(sgood(amount="abc"), sdir))
assert any("tags[0]" in v for v in srun(sgood(tags=[3]), sdir))
assert any("lines[0].account_id" in v for v in srun(sgood(lines=[{"account_id": 10}]), sdir))
assert any("lines[0].memo" in v and "not in the spec" in v for v in srun(sgood(lines=[{"memo": "x"}]), sdir))
assert any("colour" in v and "not in the spec" in v for v in srun(sgood(colour="red"), sdir))
def test_schema_derived_journal_id(sdir):
assert any("derived_id(source_document, event)" in v for v in srun(sgood(journal_id="1"), sdir))
def test_schema_path_and_query(sdir):
assert srun({"on": "2026-10-01"}, sdir, "GET", "/fx/GBP") == []
assert any("path currency" in v for v in srun({"on": "2026-10-01"}, sdir, "GET", "/fx/pound"))
assert any("on" in v and "missing" in v for v in srun({}, sdir, "GET", "/fx/GBP"))
assert any("kind" in v for v in srun({"on": "2026-10-01", "kind": "closing"}, sdir, "GET", "/fx/GBP"))
assert any("limit" in v and "not in the spec" in v for v in srun({"on": "2026-10-01", "limit": 5}, sdir, "GET", "/fx/GBP"))
def test_schema_query_array_is_comma_separated(sdir):
assert srun({"on": "2026-10-01", "ids": "1,2,3"}, sdir, "GET", "/fx/GBP") == []
assert srun({"on": "2026-10-01", "ids": [1, 2]}, sdir, "GET", "/fx/GBP") == []
assert any("ids[1]" in v for v in srun({"on": "2026-10-01", "ids": "1,x"}, sdir, "GET", "/fx/GBP"))
def test_schema_unknown_type_is_a_violation(sdir, tmp_path):
SCHEMA_SPEC["seams"][0]["request_schema"].append({"field": "z", "type": "wat", "required": False})
try:
(tmp_path / "modules" / "ledger" / "module.yaml").write_text(yaml.safe_dump(SCHEMA_SPEC))
contract._load.cache_clear()
assert any("neither a shared key nor a basic type" in v for v in srun(sgood(z=1), sdir))
finally:
SCHEMA_SPEC["seams"][0]["request_schema"].pop()
contract._load.cache_clear()
def test_a_placeholder_with_a_comma_in_a_query_is_one_value_not_a_field():
from erp_platform.contract import parse_request
names = [n for f in parse_request("query invoice_ids=<comma-separated, at most 200>, as_of=<date>", {}) for n in f.names]
assert "at" not in names and "invoice_ids" in names and "as_of" in names
def test_a_seam_shared_by_several_callers_names_each(mdir):
# 0.9.8: hr's `with: expenses, pos` — each listed caller is that seam's caller, an unlisted one is not.
from erp_platform.contract import _callers
assert _callers({"with": "expenses, pos"}) == ["expenses", "pos"]
assert _callers({"with": ["expenses", "pos"]}) == ["expenses", "pos"] and _callers({}) == []
assert run({"date": "2026-10-01"}, mdir, method="GET", path="/periods/UK01/open") == []
assert "not for seller" in check("ledger", "GET", "/periods/UK01/open", {"date": "2026-10-01"},
caller="seller", method_dir=mdir)[0]