dodil/erp-platform
tests/test_capabilities.py
"""capabilities.yaml's rules, enforced: a throwaway provider brick and a consumer, on real SQLite and real doors.
Every rule the contract states has a test that fails when the rule is broken (AGENTS.md). The claim race runs
twenty threads at one key on a FILE SQLite database — real concurrent transactions, not one shared
connection — because the property under test is "two callers, one action".
"""
import json
import socket
import threading
import time
from contextlib import contextmanager
import anyio
import httpx
import pytest
import uvicorn
from mcp import ClientSession
from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client
from sqlalchemy import create_engine, inspect, select, text
from erp_platform import capabilities, requestlog
from erp_platform.bricks import Brick
from erp_platform.capabilities import (Call, CapabilityRefused, CapabilitySeam, ConnectionRequired, Failed,
InvalidCapabilityInput, ManualTask, Result, Transient)
from erp_platform.seam import claims, gateway_headers, register_local, unregister_local
# ── a provider brick: "prov", scripted ──────────────────────────────────────────────────────
PROV = Brick("prov", instructions="A throwaway provider.")
SCRIPT: list = [] # what the next handler calls do: "done", Transient, Failed, ...
CALLS: list = [] # every handler call: (capability, key)
DELAY = {"s": 0.0}
def _next(call: Call):
CALLS.append((call.capability.name, call.idempotency_key))
if DELAY["s"]:
time.sleep(DELAY["s"])
step = SCRIPT.pop(0) if SCRIPT else "done"
if isinstance(step, Exception):
raise step
if callable(step):
return step(call)
return step
@PROV.capability("identity.provision_user", secrets=("recovery_email",))
def provision_user(call: Call):
"""Look the account up by primary_email; create it if missing."""
_next(call)
return "acct-1", {"account_id": "acct-1", "primary_email": call.input["primary_email"], "created": True}, \
"Account created."
@PROV.capability("identity.suspend_user")
def suspend_user(call: Call):
_next(call)
return "acct-1", {"account_id": "acct-1", "suspended": True}, "Suspended."
@suspend_user.plan
def _plan_suspend(call: Call) -> list:
return [{"action": "suspend the account", "target": call.input["primary_email"],
"payload": {"suspended": True}}]
@PROV.capability("drive.transfer_status")
def transfer_status(call: Call):
_next(call)
return call.input["transfer_ref"], {"transfer_ref": call.input["transfer_ref"], "state": "completed",
"transferred_items": 3}, ""
ANNA = {"given_name": "Anna", "family_name": "Berg", "primary_email": "anna@co.example",
"recovery_email": "anna.private@mail.example", "org_unit": "/Staff/London",
"initial_password_delivery": "recovery_email"}
CALLER = claims("hr-service@co.example", ["service"], ["prov:capability:invoke"])
@pytest.fixture(autouse=True)
def database(tmp_path, monkeypatch):
engine = create_engine(f"sqlite:///{tmp_path / 'prov.db'}",
connect_args={"timeout": 30, "check_same_thread": False})
capabilities.tables("prov").metadata.create_all(engine)
monkeypatch.setattr(PROV, "capability_connection", engine.begin)
monkeypatch.setattr(capabilities, "_sleep", lambda s: None)
SCRIPT.clear(), CALLS.clear()
DELAY["s"] = 0.0
capabilities.reset_providers()
yield engine
capabilities.reset_providers()
def rows(engine, table: str) -> list:
t = getattr(capabilities.tables("prov"), table)
with engine.connect() as c:
return [dict(r._mapping) for r in c.execute(select(t))]
# ── the contract ────────────────────────────────────────────────────────────────────────────
def test_the_contract_loads_typed_and_the_result_type_is_its_result_shape():
k = capabilities.contract()
assert k.version == "0.6" and len(k.capabilities) == 22
assert tuple(k.result) == tuple(Result.__dataclass_fields__)
cap = k.get("identity.provision_user")
assert [f.name for f in cap.input][:3] == ["given_name", "family_name", "primary_email"]
assert cap.tool_name == "identity_provision_user" and cap.route == "/capabilities/identity.provision_user"
assert k.get("identity.delete_user").destructive and k.get("drive.transfer_status").read
with pytest.raises(capabilities.UnknownCapability, match="not a capability"):
k.get("identity.provision_robot")
def test_the_0_4_entries_load_with_their_new_type_forms():
"""identity.invite_app_user and contracts.* (0.4): one object, a map of scalars, UPPER_CASE codes, a literal."""
k = capabilities.contract()
draft = k.get("contracts.draft")
ok = draft.validate_input({
"type_code": "FREELANCE_SERVICES", "counterparty_bp_id": 812, "context_ref": None,
"source_ref": "hr:onboarding/r-1|3", "variables": {"day_rate": "450.00", "days": 20, "remote": True},
"signers": [{"side": "THEM", "full_name": "Kim Lee", "email": "kim@client.example", "title": None,
"authority_basis": None, "signing_order": 1}]})
assert ok["signers"][0]["side"] == "THEM" and ok["variables"]["days"] == 20
for bad in ({"signers": [{**ok["signers"][0], "side": "BOTH"}]}, {"variables": {"nested": {"no": 1}}},
{"counterparty_bp_id": "812"}):
with pytest.raises(InvalidCapabilityInput):
draft.validate_input({**ok, **bad})
send = k.get("contracts.send")
assert send.validate_output({"contract_id": "c", "state": "OUT_FOR_SIGNATURE", "version": 1, "sha256": "ab",
"deliveries": [{"signer_no": 1, "channel": "manual"}]})["state"] == "OUT_FOR_SIGNATURE"
with pytest.raises(InvalidCapabilityInput):
send.validate_output({"contract_id": "c", "state": "DRAFT", "version": 1, "sha256": "ab", "deliveries": []})
status = k.get("contracts.status")
assert status.read and k.get("contracts.fetch_executed").read and not k.get("contracts.draft").read
out = status.validate_output({"contract_id": "c", "state": "SIGNED", "version": 2, "signers": [],
"executed": None, "key_dates": [], "principal_statement_issued_at": None,
"warnings": []})
assert out["executed"] is None
invite = k.get("identity.invite_app_user")
assert invite.tool_name == "identity_invite_app_user" and not invite.destructive
assert [f.name for f in invite.output] == ["sub", "email", "created"]
def test_0_6_identity_revoke_app_user_has_hr_s_shape_and_is_a_manual_task_without_a_provider(monkeypatch):
"""0.6: moved from brick-hr's capabilities.local.yaml, same name and shape; rule 8 holds for it too."""
k = capabilities.contract()
revoke = k.get("identity.revoke_app_user")
assert revoke.tool_name == "identity_revoke_app_user" and not revoke.destructive and not revoke.read
assert [f.name for f in revoke.input] == ["pool", "email", "sub", "reason"]
assert [f.name for f in revoke.output] == ["disabled", "sessions_revoked"]
monkeypatch.delenv("CAPABILITY_IDENTITY_REVOKE_APP_USER", raising=False)
task = CapabilitySeam("identity.revoke_app_user").call(
"revoke:p-1", {"pool": "acme/staff", "email": "kim@co.example", "reason": "leaver"}, actor="hr@co.example")
assert isinstance(task, ManualTask) and "no provider" in task.reason
assert task.input == {"pool": "acme/staff", "email": "kim@co.example", "sub": None, "reason": "leaver"}
assert "kim@co.example" in task.plan and "acme/staff" in task.plan
with pytest.raises(InvalidCapabilityInput):
revoke.validate_input({"pool": "acme/staff", "reason": "leaver"})
def test_input_is_validated_strictly_against_the_contract():
cap = capabilities.contract().get("drive.create_folder")
ok = cap.validate_input({"name": "Anna", "share_with": [{"email": "a@co.example", "role": "writer"}]})
assert ok == {"name": "Anna", "parent": None, "share_with": [{"email": "a@co.example", "role": "writer"}]}
for bad in ({"name": "Anna"}, # a required list missing
{"name": "", "share_with": []}, # an empty required string
{"name": "A", "share_with": [{"email": "a@co.example", "role": "admin"}]}, # not a choice
{"name": "A", "share_with": [], "colour": "red"}, # not in the contract
{"name": 7, "share_with": []}): # not coerced
with pytest.raises(InvalidCapabilityInput):
cap.validate_input(bad)
with pytest.raises(InvalidCapabilityInput, match="ISO"):
capabilities.contract().get("mail.set_autoreply").validate_input(
{"mailbox": "a@co.example", "subject": "s", "body_text": "b", "start": "next tuesday"})
def test_the_ddl_providers_ship_is_the_schema_the_code_uses(tmp_path, monkeypatch):
"""Split and classified by erp-migrate itself, so what a provider pastes into schema.sql migrates."""
monkeypatch.setenv("BUCKET", "not-used")
from erp_platform import migrate
engine = create_engine(f"sqlite:///{tmp_path / 'ddl.db'}")
stmts = list(migrate.statements(capabilities.ddl("prov")))
assert len(stmts) == 2 and {migrate.classify(s) for s in stmts} == {"SAFE"}
with engine.begin() as c:
for stmt in stmts:
c.execute(text(stmt))
t = capabilities.tables("prov")
assert migrate.expected_columns(capabilities.ddl("prov")) == {
t.claims.name: list(t.claims.c.keys()), t.log.name: list(t.log.c.keys())}
insp = inspect(engine)
for table in (t.claims, t.log):
assert {c["name"] for c in insp.get_columns(table.name)} == set(table.c.keys())
assert insp.get_pk_constraint(table.name)["constrained_columns"] == [c.name for c in table.primary_key]
# ── rule 1: claim, replay, in progress, re-attempt, release, 409 ────────────────────────────
def test_a_key_acts_once_and_replays_the_stored_result(database):
first = provision_user.invoke(CALLER, "run-1|3", ANNA)
again = provision_user.invoke(CALLER, "run-1|3", ANNA)
assert first.status == "done" and not first.idempotent_replay and first.provider == "prov"
assert again.status == "done" and again.idempotent_replay and again.output == first.output
assert len(CALLS) == 1
assert [r["generation"] for r in rows(database, "claims")] == [1]
def test_an_unexpired_claim_with_no_result_answers_in_progress():
seen = {}
def nested(call):
seen["r"] = provision_user.invoke(CALLER, "run-1|3", ANNA) # the same key, while this one runs
return "done"
SCRIPT.append(nested)
assert provision_user.invoke(CALLER, "run-1|3", ANNA).status == "done"
assert seen["r"].status == "in_progress" and len(CALLS) == 1
def test_an_expired_claim_is_re_attempted_under_the_next_generation(database):
"""The pod died between claim and result: its claim had a short lease, and lapses."""
cap = capabilities.contract().get("identity.provision_user")
store = provision_user.store
digest = cap.input_hash(cap.validate_input(ANNA))
assert store.insert_claim(cap.name, "run-1|3", 1, digest, "dead-pod", lease=0.05)
assert provision_user.invoke(CALLER, "run-1|3", ANNA).status == "in_progress" and not CALLS
time.sleep(0.1)
r = provision_user.invoke(CALLER, "run-1|3", ANNA)
assert r.status == "done" and len(CALLS) == 1
claims_ = sorted(rows(database, "claims"), key=lambda r: r["generation"])
assert [(c["generation"], c["status"]) for c in claims_] == [(1, None), (2, "done")]
assert provision_user.invoke(CALLER, "run-1|3", ANNA).idempotent_replay
@pytest.mark.parametrize("outcome, status", [(Transient("429 from the vendor"), "failed"),
(ConnectionRequired("no Workspace connection"), "connection_required")])
def test_a_non_terminal_outcome_releases_the_claim_so_a_retry_acts(database, outcome, status):
SCRIPT.extend([outcome] * (3 if isinstance(outcome, Transient) else 1))
r = provision_user.invoke(CALLER, "run-1|3", ANNA)
assert r.status == status and (r.retryable if status == "failed" else True)
assert rows(database, "claims") == []
again = provision_user.invoke(CALLER, "run-1|3", ANNA)
assert again.status == "done" and not again.idempotent_replay
def test_a_transient_failure_is_retried_inside_the_call_before_it_is_reported():
SCRIPT.extend([Transient("timeout"), Transient("503")])
r = provision_user.invoke(CALLER, "run-1|3", ANNA)
assert r.status == "done" and len(CALLS) == 3
def test_a_permanent_failure_is_terminal_and_replayed(database):
SCRIPT.append(Failed("the domain co.example is not verified in this Workspace"))
r = provision_user.invoke(CALLER, "run-1|3", ANNA)
assert (r.status, r.retryable) == ("failed", False)
again = provision_user.invoke(CALLER, "run-1|3", ANNA)
assert again.idempotent_replay and again.message == r.message and len(CALLS) == 1
def test_the_same_key_with_different_input_is_409():
from erp_platform.bricks import Refused
provision_user.invoke(CALLER, "run-1|3", ANNA)
with pytest.raises(Refused) as e:
provision_user.invoke(CALLER, "run-1|3", {**ANNA, "org_unit": "/Staff/Paris"})
assert e.value.status == 409 and "different input" in e.value.reason
def test_a_crash_releases_the_claim_and_is_logged_by_type(database):
SCRIPT.append(RuntimeError("bug with anna@co.example in it"))
with pytest.raises(RuntimeError):
provision_user.invoke(CALLER, "run-1|3", ANNA)
assert rows(database, "claims") == []
[entry] = rows(database, "log")
assert entry["status"] == "error" and entry["error"] == "RuntimeError" and entry["message"] is None
def test_concurrent_calls_with_one_key_act_exactly_once(database):
DELAY["s"] = 0.3
results, errors = [], []
def go():
try:
results.append(provision_user.invoke(CALLER, "run-1|3", ANNA))
except Exception as exc: # noqa: BLE001 — reported below
errors.append(exc)
threads = [threading.Thread(target=go) for _ in range(20)]
for t in threads:
t.start()
for t in threads:
t.join(30)
assert not errors, errors
assert len(CALLS) == 1, f"the handler acted {len(CALLS)} times for one key"
fresh = [r for r in results if r.status == "done" and not r.idempotent_replay]
assert len(fresh) == 1 and {r.status for r in results} <= {"done", "in_progress"}
# ── rule 2: dry run; rule 5: the action log; reads ──────────────────────────────────────────
def test_a_dry_run_plans_and_claims_nothing(database):
r = provision_user.invoke(CALLER, "run-1|3", ANNA, dry_run=True)
assert r.status == "planned" and r.plan[0]["payload"]["recovery_email"] == "[redacted]"
assert rows(database, "claims") == [] and not CALLS
s = suspend_user.invoke(CALLER, "run-1|9", {"primary_email": "anna@co.example", "reason": "left"}, dry_run=True)
assert s.plan == [{"action": "suspend the account", "target": "anna@co.example", "payload": {"suspended": True}}]
assert provision_user.invoke(CALLER, "run-1|3", ANNA).status == "done" # the real call still acts
def test_every_call_is_in_the_action_log_with_secrets_redacted(database):
provision_user.invoke(CALLER, "run-1|3", ANNA, dry_run=True)
provision_user.invoke(CALLER, "run-1|3", ANNA)
provision_user.invoke(CALLER, "run-1|3", ANNA)
log = sorted(rows(database, "log"), key=lambda r: r["event_at"])
assert [(r["status"], r["dry_run"], r["replay"]) for r in log] == [
("planned", True, False), ("done", False, False), ("done", False, True)]
assert all(json.loads(r["input"])["recovery_email"] == "[redacted]" for r in log)
assert all("anna.private" not in json.dumps(r, default=str) for r in log)
assert log[1]["actor"] == "hr-service@co.example" and log[1]["provider_ref"] == "acct-1" and log[1]["tries"] == 1
def test_a_read_capability_takes_no_claim(database):
for _ in range(2):
r = transfer_status.invoke(CALLER, "any", {"transfer_ref": "t-9"})
assert r.status == "done" and not r.idempotent_replay
assert len(CALLS) == 2 and rows(database, "claims") == []
def test_a_provider_output_that_breaks_the_contract_is_a_bug_not_a_result(database):
SCRIPT.append(lambda call: None)
bad = Brick("bad")
@bad.capability("groups.add_member")
def add(call):
return "g", {"group_id": "g"}, "" # `added` missing
bad.capability_connection = PROV.capability_connection
capabilities.tables("bad").metadata.create_all(database)
with pytest.raises(InvalidCapabilityInput, match="added"):
add.invoke(CALLER, "k", {"group": "staff", "member_email": "a@co.example", "role": "member"})
def test_a_secret_must_be_an_input_of_the_capability():
with pytest.raises(ValueError, match="not input fields"):
Brick("x").capability("mail.send", secrets=("password",))(lambda call: None)
def test_the_capability_door_asks_for_the_spec_namespace_not_the_brick_name():
# 0.9.8: the brick named contracts has the namespace `contract`; its door and its consumers agree on it.
named = Brick("contracts-x", namespace="contract-x")
named.capability("groups.add_member")(lambda call: ("g", {"group_id": "g", "added": True}, ""))
assert named.use_cases[-1].permission == "contract-x:capability:invoke"
assert capabilities.invoke_permission("contracts-x") == "contract-x:capability:invoke"
assert capabilities.invoke_permission("unregistered") == "unregistered:capability:invoke"
assert Brick("plain").namespace == "plain"
# ── the doors: route, permission, MCP tool name and hints ───────────────────────────────────
@pytest.fixture(scope="module")
def server():
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
port = s.getsockname()[1]
srv = uvicorn.Server(uvicorn.Config(PROV.app(), host="127.0.0.1", port=port, log_level="critical"))
thread = threading.Thread(target=srv.run, daemon=True)
thread.start()
deadline = time.time() + 10
while not srv.started and time.time() < deadline:
time.sleep(0.02)
yield f"http://127.0.0.1:{port}"
srv.should_exit = True
thread.join(5)
def test_the_route_is_the_contract_and_needs_the_providers_permission(server):
body = {"idempotency_key": "run-1|3", "dry_run": False, "input": ANNA}
r = httpx.post(f"{server}/capabilities/identity.provision_user", json=body, headers=gateway_headers(CALLER))
assert r.status_code == 200 and r.json()["status"] == "done" and r.json()["provider"] == "prov"
nobody = gateway_headers(claims("x@co.example", ["service"], []))
r = httpx.post(f"{server}/capabilities/identity.provision_user", json=body, headers=nobody)
assert r.status_code == 403 and "prov:capability:invoke" in r.json()["detail"]
r = httpx.post(f"{server}/capabilities/identity.provision_user", headers=gateway_headers(CALLER),
json={**body, "input": {**ANNA, "initial_password_delivery": "sms"}})
assert r.status_code == 422
async def _tools(base, headers, call=None):
async with create_mcp_http_client(headers=headers) as http:
async with streamable_http_client(f"{base}/mcp", http_client=http) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
listed = {t.name: t for t in (await session.list_tools()).tools}
result = await session.call_tool(*call) if call else None
return listed, result
def test_the_mcp_tool_is_the_name_with_underscores_and_carries_the_contracts_hints(server):
listed, result = anyio.run(_tools, server, gateway_headers(CALLER),
("identity_provision_user", {"idempotency_key": "mcp-1", "input": ANNA}))
assert set(listed) == {"identity_provision_user", "identity_suspend_user", "drive_transfer_status"}
assert listed["identity_suspend_user"].annotations.destructive_hint is True
assert listed["identity_provision_user"].annotations.destructive_hint is False
assert listed["drive_transfer_status"].annotations.read_only_hint is True
assert not result.is_error and json.loads(result.content[0].text)["status"] == "done"
# ── consumer side ───────────────────────────────────────────────────────────────────────────
PROVISION = CapabilitySeam("identity.provision_user")
def test_with_no_provider_the_call_is_a_manual_task_with_the_exact_input(monkeypatch):
monkeypatch.delenv("CAPABILITY_IDENTITY_PROVISION_USER", raising=False)
minimal = {k: v for k, v in ANNA.items() if k != "recovery_email"}
task = PROVISION.call("run-1|3", minimal, actor="hr@co.example")
assert isinstance(task, ManualTask) and task.status == "manual"
assert task.input == {**minimal, "recovery_email": None} # validated, normalised
assert task.capability == "identity.provision_user" and "no provider" in task.reason
assert "anna@co.example" in task.plan and "/Staff/London" in task.plan
stored = ManualTask.from_row(json.loads(json.dumps(task.as_row())))
assert stored == task
done = stored.complete(provider_ref="C0123")
assert (done.status, done.provider, done.provider_ref) == ("done", "manual", "C0123")
assert not CALLS
def test_a_consumer_bug_fails_before_anything_is_sent():
with pytest.raises(InvalidCapabilityInput, match="primary_email"):
PROVISION.call("run-1|3", {"given_name": "Anna"})
with pytest.raises(capabilities.UnknownCapability):
CapabilitySeam("identity.provision_robot")
@contextmanager
def in_process():
register_local("prov", PROV.routes_app())
capabilities.register_provider("identity.provision_user", "prov")
try:
yield
finally:
unregister_local("prov")
def test_tier_one_reaches_the_provider_in_process_with_its_permission_and_the_request_id(database):
with in_process():
token = requestlog._CURRENT.set(requestlog._Request("req-42"))
try:
r = PROVISION.call("run-1|3", ANNA, actor="hr@co.example")
finally:
requestlog._CURRENT.reset(token)
assert isinstance(r, Result) and r.status == "done" and r.provider == "prov"
[entry] = rows(database, "log")
assert entry["request_id"] == "req-42" and entry["actor"] == "hr@co.example"
def test_connection_required_becomes_a_manual_task_naming_what_is_missing():
SCRIPT.append(ConnectionRequired("connect Google Workspace in Settings → Connections"))
with in_process():
task = PROVISION.call("run-1|3", ANNA)
assert isinstance(task, ManualTask) and "connect Google Workspace" in task.reason
def test_tier_two_names_the_provider_in_an_env_var_and_finds_it_at_its_base_url(server, monkeypatch):
monkeypatch.setenv("CAPABILITY_IDENTITY_PROVISION_USER", "prov")
monkeypatch.setenv("PROV_BASE_URL", server)
r = PROVISION.call("tier2-1", ANNA)
assert isinstance(r, Result) and r.status == "done"
monkeypatch.setenv("CAPABILITY_IDENTITY_PROVISION_USER", "manual")
assert isinstance(PROVISION.call("tier2-1", ANNA), ManualTask)
def test_a_provider_that_is_down_is_a_retryable_failure_and_a_4xx_is_the_consumers_bug(server, monkeypatch):
monkeypatch.setenv("CAPABILITY_IDENTITY_PROVISION_USER", "prov")
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
closed = s.getsockname()[1]
monkeypatch.setenv("PROV_BASE_URL", f"http://127.0.0.1:{closed}")
r = PROVISION.call("down-1", ANNA)
assert (r.status, r.retryable) == ("failed", True) and "unreachable" in r.message
monkeypatch.setenv("PROV_BASE_URL", server)
PROVISION.call("dup-1", ANNA)
with pytest.raises(CapabilityRefused) as e:
PROVISION.call("dup-1", {**ANNA, "org_unit": "/Elsewhere"})
assert e.value.status == 409