dodil/erp-brick-contracts
tests/test_tier1.py
"""Tier 1 with the REAL neighbours: hr, contracts and business-partner composed as ONE app on ONE bucket (one SQLite
engine here). hr asks for its starter's contract through the contracts.* capabilities exactly as its
infrastructure/contracts_acl.py does (source_ref hr:<run_id>#<step_no>, key <source_ref>|<type_code>, the person's
bp_id, context person_id|n, scalar values, signers only when the director party is set); contracts reads the
counterparty through business-partner's route; the starter signs through the public door and the director inside the
app; hr's tick then polls contracts.status and, as the ORIGINATOR, fetches the executed copy and files it on the
person — the file hr keeps has the sha256 contracts gave it.
"""
import base64
import hashlib
import json
import pathlib
from contextlib import ExitStack
import anyio
import httpx
import pytest
import yaml
from sqlalchemy import create_engine, event
from erp_platform import compose, seam, work
from erp_platform.scenarios import Catalogue
from erp_platform.scenarios.inprocess import stage
from seed import seed as seed_contracts
HERE = pathlib.Path(__file__).parent
ROLES = yaml.safe_load((HERE / "tier1.roles.yaml").read_text())["roles"]
ORG, SINCE, DOMAIN = "acme", "2020-01-01", "co.example"
def who(role, n=1, sub=None):
s = sub or f"{role}.{n}"
return seam.gateway_headers({"sub": s, "email": f"{s}@scenario.test", "app_roles": [role],
"permissions": ROLES[role]})
HR, ADMIN, ADMIN2, DIRECTOR, TICK = who("hr_admin"), who("contract_admin"), who("contract_admin", 2), who("director"), \
who("clock")
@pytest.fixture
def world(tmp_path):
e = create_engine(f"sqlite:///{tmp_path / 'bucket.db'}", connect_args={"check_same_thread": False, "timeout": 2})
@event.listens_for(e, "connect")
def _wal(conn, _rec):
conn.execute("PRAGMA journal_mode=WAL")
cat = Catalogue("tier1", compose=HERE / "tier1.compose.yaml", roles=HERE / "tier1.roles.yaml")
from brick_contracts.infrastructure import repositories
repositories.forget_snapshots()
with ExitStack() as st:
staged = stage(cat, st, lambda name: e) # every brick and the link tables: one bucket
st.callback(work.reset_clock)
yield staged
e.dispose()
def call(app, method, path, h, body=None, form=None, base="http://staff"):
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url=base) as c:
if form is not None:
return await c.request(method, path, headers=h, data=form)
return await c.request(method, path, headers=h, json=body)
r = anyio.run(go)
try:
return r.status_code, r.json()
except ValueError:
return r.status_code, r.text
def at(when):
work.set_clock(at=work.parse_instant(f"{when} Europe/London"), freeze=True)
def ok(app, method, path, h, body=None):
s, out = call(app, method, path, h, body)
assert 200 <= s < 300, (path, s, out)
return out
def seed_hr(app):
post = lambda p, b: ok(app, "POST", "/hr" + p, HR, b) # noqa: E731
post("/localisation-packs/gb", {"with_playbooks": True})
post("/configuration/settings", {"rows": [{"key": "email_domain", "value": DOMAIN},
{"key": "default_org_id", "value": ORG},
{"key": "default_legal_entity", "value": "ACME-UK"},
{"key": "hr_mailbox", "value": f"people@{DOMAIN}"}]})
post("/configuration/job", {"rows": [{"job_code": "designer", "title": "Designer"},
{"job_code": "studio_lead", "title": "Studio Lead"}]})
post("/configuration/location", {"rows": [{"location_code": "LON", "name": "London", "address": "1 Example Sq",
"country": "GB", "timezone": "Europe/London",
"holiday_calendar": "gb-eaw"}]})
post("/configuration/work_schedule", {"rows": [{"schedule_code": "FT", "label": "37.5 h", "weekly_minutes": 2250,
"full_time_weekly_minutes": 2250,
"working_weekdays": [0, 1, 2, 3, 4]}]})
post("/org", {"org_id": ORG, "kind": "org", "values": {"name": "Acme", "legal_entity_ref": "ACME-UK"}})
post("/org", {"org_id": ORG, "kind": "unit", "id": "STUDIO", "valid_from": SINCE,
"values": {"name": "Studio", "location_code": "LON"}})
def hire(app, given, family, start, **extra):
return ok(app, "POST", "/hr/hires", HR, {
"legal_given_name": given, "legal_family_name": family, "worker_type": "employee_ft", "job_code": "designer",
"org_unit_id": "STUDIO", "start_date": start, "location_code": "LON", "schedule_code": "FT",
"personal_email": f"{given.lower()}.{family.lower()}@home.example", "rate_amount": "42000.0000",
"rate_basis": "year", "currency": "GBP", **extra})
def director_person(app):
"""A director in hr with a work email the pool knows (director.1@scenario.test): the run's signing director."""
at("2024-01-10T10:00")
d = hire(app, "Dana", "Director", "2024-01-08", job="studio_lead")
at("2026-10-19T10:00")
p = ok(app, "GET", f"/hr/people/{d['person_id']}", HR)
first = p["detail_history"][0]["valid_from"]
ok(app, "POST", f"/hr/people/{d['person_id']}/corrections", HR,
{"record": "detail", "valid_from": first, "values": {"work_email": "director.1@scenario.test"}})
return d["person_id"]
def contract_step(app, run_id):
r = ok(app, "GET", f"/hr/runs/{run_id}", HR)
return next(s for s in r["steps"] if s.get("kind") == "document" or s["step_no"] == "3")
def cid_of(step):
return (step.get("result") or {}).get("contract_id") or step.get("contract_id")
def test_hr_drafts_sends_and_files_the_signed_employment_contract(world):
app, door = world.app, world.public_app
at("2026-10-19T10:00")
seed_hr(app)
seed_contracts(lambda p, b: call(app, "POST", p, ADMIN, b), lambda p, b: call(app, "POST", p, ADMIN2, b),
prefix="/contracts")
director = director_person(app)
h = hire(app, "Anna", "Berg", "2026-11-02")
rid = h["run_id"]
ok(app, "POST", f"/hr/runs/{rid}/parties", HR, {"group_code": "director", "person_id": director})
# The partner hr synced has NO role until the start date (hr gives it then): the type allows a role-less person.
s, p = call(app, "GET", f"/business-partner/partners/{h['bp_id']}", HR)
assert s == 200 and p["roles"] == [], p
ok(app, "POST", "/hr/run-due-steps", TICK, {}) # hr: contracts.draft, then contracts.send
step = contract_step(app, rid)
cid = cid_of(step)
assert cid and cid.startswith("ct-"), step
c = ok(app, "GET", f"/contracts/contracts/{cid}", ADMIN)
assert c["stored_state"] == "OUT_FOR_SIGNATURE" and c["source_ref"] == f"hr:{rid}#3", c
assert c["owner_group"] == "hr_admin" and c["sensitivity"] == "employment"
assert {v["name"]: v["value"] for v in c["variables"]}["legal_name"] == "Anna Berg"
sides = {x["side"]: x for x in c["signers"]}
assert sides["US"]["email"] == "director.1@scenario.test" and sides["US"]["kind"] == "INTERNAL"
assert sides["THEM"]["email"] == "anna.berg@home.example" and sides["THEM"]["kind"] == "EXTERNAL"
sha = c["versions"][-1]["sha256"]
out = ok(app, "POST", f"/contracts/contracts/{cid}/sign", DIRECTOR, {"version": 1, "sha256": sha,
"typed_name": "Dana Director", "consent": True})
assert out["status"] == "SIGNED", out
them = sides["THEM"]["signer_no"]
url = ok(app, "POST", f"/contracts/contracts/{cid}/signers/{them}/copy-link", HR, {})["url"]
path = url.split("://", 1)[1].split("/", 1)[1]
s, page = call(door, "POST", "/" + path, {}, form={"action": "sign", "typed_name": "Anna Berg", "consent": "yes",
"version": "1", "sha256": sha}, base=world.public_base)
assert s == 200 and "signed electronically" in page, page[:300]
executed = ok(app, "GET", f"/contracts/contracts/{cid}", ADMIN)["executed"]
assert executed and executed["provenance"] == "GENERATED"
at("2026-10-19T12:00") # past hr's poll interval
ok(app, "POST", "/hr/run-due-steps", TICK, {}) # hr: contracts.status → fetch_executed
step = contract_step(app, rid)
assert step["state"] == "done", step
docs = ok(app, "GET", f"/hr/people/{h['person_id']}/documents", HR)["documents"]
mine = [d for d in docs if d.get("contract_id") == cid or str(d.get("doc_id", "")).endswith(cid)]
assert mine and mine[0]["sha256"] == executed["sha256"], docs
def test_another_consumer_never_gets_hrs_executed_copy(world):
"""crm's identity (another consumer) asking for hr's contract: status without `executed`, fetch 404."""
app = world.app
at("2026-10-19T10:00")
seed_hr(app)
seed_contracts(lambda p, b: call(app, "POST", p, ADMIN, b), lambda p, b: call(app, "POST", p, ADMIN2, b),
prefix="/contracts")
h = hire(app, "Ben", "Cole", "2026-11-02")
ok(app, "POST", "/hr/run-due-steps", TICK, {})
cid = cid_of(contract_step(app, h["run_id"]))
crm = seam.gateway_headers({"sub": "crm-adapter", "email": "crm-adapter", "app_roles": ["service"],
"permissions": ["contract:capability:invoke"]})
s, st = call(app, "POST", "/contracts/capabilities/contracts.status", crm,
{"idempotency_key": f"s|{cid}", "input": {"contract_id": cid}})
assert s == 200 and st["output"]["executed"] is None, st
s, _ = call(app, "POST", "/contracts/capabilities/contracts.fetch_executed", crm,
{"idempotency_key": f"f|{cid}", "input": {"contract_id": cid}})
assert s == 404
def test_a_contract_that_needed_approval_reaches_hr_once_approved(world):
app = world.app
at("2026-10-19T10:00")
seed_hr(app)
seed_contracts(lambda p, b: call(app, "POST", p, ADMIN, b), lambda p, b: call(app, "POST", p, ADMIN2, b),
prefix="/contracts")
# A standard clause rewritten in the employment template makes every draft of it need legal's approval.
ok(app, "POST", "/contracts/clauses", ADMIN, {"clause_id": "employment_terms_fb", "family": "employment",
"position": "FALLBACK", "deviation_level": 1, "title": "Employment",
"body": "{{legal_name}} is employed as {{job_title}}."})
from seed import employment_template
t = employment_template()
t["slots"][0]["clause_id"] = "employment_terms_fb"
v = ok(app, "POST", "/contracts/templates", ADMIN, t)["version"]
ok(app, "POST", f"/contracts/templates/employment-std/versions/{v}/approve", ADMIN2, {})
director = director_person(app)
h = hire(app, "Cy", "Dee", "2026-11-02")
ok(app, "POST", f"/hr/runs/{h['run_id']}/parties", HR, {"group_code": "director", "person_id": director})
ok(app, "POST", "/hr/run-due-steps", TICK, {})
cid = cid_of(contract_step(app, h["run_id"]))
legal = seam.gateway_headers({"sub": "legal.1", "email": "legal.1@scenario.test", "app_roles": ["legal"],
"permissions": ["contract:approval:decide", "contract:contract:read"]})
c = ok(app, "GET", f"/contracts/contracts/{cid}", ADMIN)
assert c["stored_state"] == "IN_APPROVAL"
ok(app, "POST", f"/contracts/contracts/{cid}/approvals/{c['approvals'][-1]['seq']}/decide", legal,
{"decision": "approve"})
for _ in range(3):
at("2026-10-19T12:00")
ok(app, "POST", "/hr/run-due-steps", TICK, {})
assert ok(app, "GET", f"/contracts/contracts/{cid}", ADMIN)["stored_state"] == "OUT_FOR_SIGNATURE"
def test_the_three_bricks_compose_without_collisions():
report = compose.check(compose.load(HERE / "tier1.compose.yaml"))
assert report["problems"] == [], report["problems"]
wired = {c["capability"]: c["provider"] for c in report["capabilities"]}
assert {wired[k] for k in ("contracts.draft", "contracts.send", "contracts.status", "contracts.fetch_executed")} \
== {"contracts"}
assert wired["mail.send"] == "manual"