"""contracts' scenario catalogue: SC-CT-01 … SC-CT-25 (modules/contracts tests.scenarios), each mapped to its stories,
written ONCE over the composition's doors — routes, MCP tools, the public link — never a table (SC-CT-11 alone writes
one row on purpose: it IS "someone with bucket access edits an event", in process only).
erp-scenarios list tests/scenarios/catalogue.py
erp-scenarios run tests/scenarios/catalogue.py # in process: contracts + business-partner
Every scenario makes its own partners (s.key) and, where it changes configuration, its own contract type and
template, so none leans on another's rows or order (--shuffle). No mail provider: every delivery is a manual task.
LIVE: run against a demo deployment; the clock scenarios need the demo clock (CT_DEMO_CLOCK=enabled) and are skipped
without it; SC-CT-11 is in process only.
"""
import pathlib
import sys
import uuid
HERE = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent))
from erp_platform.scenarios import Catalogue, Rows # noqa: E402
from seed import DIRECTOR, employment_template, freelance_template, seed # noqa: E402
CAT = Catalogue("contracts", compose=HERE / "compose.yaml", roles=HERE / "roles.yaml", ticks=["POST /contracts/run-due"],
tick_as="clock", clock="contracts", zone="Europe/London")
C = "/contracts"
def uniq() -> str:
return uuid.uuid4().hex[:8]
# ── helpers (all through the doors) ────────────────────────────────────────────────────────
def admin(s, n=1):
"""contract_admin — reached directly, so a generated probe that declares only the personas it tests (the rows
probe: owner and owner#2) can still be given its standard paper and its partner."""
from erp_platform.scenarios.core import Actor
return Actor(s, "contract_admin" if n == 1 else f"contract_admin#{n}")
def setup(s):
def call(actor):
def go(p, b):
r = actor.post(p, b)
return r.status_code, r.body
return go
seed(call(admin(s)), call(admin(s, 2)), prefix=C)
def org(s, name="Studio Nine", roles=("supplier",)):
legal = f"{name} {uniq()} Ltd"
r = admin(s).put("/business-partner/partners", {"bp_key": s.key("BP").lower(), "legal_name": legal, "country": "GB",
"roles": list(roles)}).ok()
return str(r.body["bp_id"]), legal
def person(s, name="Sam Starter", roles=()):
pid = str(int(uuid.uuid4().int % 10 ** 12))
r = s.as_("hr_service").post("/business-partner/partners/persons/sync", {"person": {"person_id": pid, "legal_name": name,
"roles": list(roles)}}).ok()
return str(r.body["bp_id"])
def signers(s, them="Sam Nine", order_them=1, order_us=2, us_email=None, extra_them=None):
out = [{"side": "THEM", "full_name": them, "email": s.email(them.split()[0].lower()), "signing_order": order_them},
{"side": "US", "full_name": "Dana Director", "email": us_email or DIRECTOR["email"], "title": "Director",
"signing_order": order_us}]
if extra_them:
out.append({"side": "THEM", "full_name": extra_them, "email": s.email("second"), "signing_order": order_them})
return out
def nda(s, who=("owner", 1), bp=None, sig=None, **values):
bp = bp or org(s)[0]
body = {"type_code": "nda", "bp_id": bp, "form_key": s.key("FK"),
"values": {"start_date": "2026-11-01", **values}, "signers": sig if sig is not None else signers(s)}
return s.as_(*who).post(f"{C}/contracts", body).ok().body
def send(s, cid, who=("owner", 1)):
return s.as_(*who).post(f"{C}/contracts/{cid}/send", {})
def copy_link(s, cid, signer_no=1, who=("owner", 1), which="sign"):
return s.as_(*who).post(f"{C}/contracts/{cid}/signers/{signer_no}/copy-link", {"which": which}).ok().body["url"]
def sign_link(s, url, sha, version=1, name="Sam Nine"):
return s.link(url, "the signer").submit({"action": "sign", "typed_name": name, "consent": "yes",
"version": str(version), "sha256": sha})
def sign_us(s, cid, sha, version=1, n=1):
return s.as_("director", n).post(f"{C}/contracts/{cid}/sign", {"version": version, "sha256": sha,
"typed_name": "Dana Director", "consent": True})
def show(s, cid, who=("contract_admin", 1)):
return s.as_(*who).get(f"{C}/contracts/{cid}")
def signed_nda(s, **values):
c = nda(s, **values)
out = send(s, c["contract_id"]).ok().body
sign_link(s, copy_link(s, c["contract_id"]), out["sha256"]).status(200)
sign_us(s, c["contract_id"], out["sha256"]).ok()
return c["contract_id"], out
def own_type(s, base: str, template: dict, **changes):
"""A contract type and template of this scenario's own (configuration it changes stays its own)."""
code = f"{base}_{uniq()}"[:60]
row = {"type_code": code, "label": changes.pop("label", base.replace("_", " ")), "retention_rule": "simple-6y",
"approval_policy": "standard", **changes}
admin(s).post(f"{C}/types", row).ok()
tmpl = {**template, "template_id": f"tpl-{code}"[:64], "type_code": code}
v = admin(s).post(f"{C}/templates", tmpl).ok().body["version"]
admin(s, 2).post(f"{C}/templates/{tmpl['template_id']}/versions/{v}/approve", {}).ok()
return code
def tasks(c):
return [t for t in c["tasks"]]
def cap(s, who, name, key, inp):
return s.as_(who).post(f"{C}/capabilities/{name}", {"idempotency_key": key, "dry_run": False, "input": inp})
# ── SC-CT-01 … 04: approval ────────────────────────────────────────────────────────────────
@CAT.scenario("SC-CT-01", "a clause deviation needs approval", stories=["CT-06", "CT-07", "CT-08"])
def deviation_needs_approval(s):
setup(s)
s.given("a draft with the Preferred liability clause and no deviations")
c = nda(s)
cid = c["contract_id"]
s.check(c["state"] == "DRAFT" and not c["approval_required"], "a standard draft needs no approval", c)
s.when("the owner switches it to the Fallback position and tries to send")
c = s.as_("owner").post(f"{C}/contracts/{cid}/sections/2/position", {"clause_id": "liability_fb"}).ok().body
s.check(c["state"] == "IN_APPROVAL" and c["sections"][1]["deviation_level"] == 1, "now in approval", c)
r = send(s, cid).refused(409)
s.then('send is refused ("approval by legal needed … liability cap is a Fallback")')
s.check("approval by legal needed" in r.reason and "liability cap is a Fallback" in r.reason, "the reason", r.reason)
q = s.as_("legal").get(f"{C}/approvals").ok().body["approvals"]
mine = [a for a in q if a["contract_id"] == cid]
s.check(len(mine) == 1 and mine[0]["draft_fingerprint"] == c["draft_fingerprint"], "in legal's queue, bound to F", q)
s.as_("legal").post(f"{C}/contracts/{cid}/approvals/{mine[0]['seq']}/decide", {"decision": "approve"}).ok()
send(s, cid).ok().field("state", "OUT_FOR_SIGNATURE")
@CAT.scenario("SC-CT-02", "an edit after approval voids it", stories=["CT-07"])
def edit_voids_approval(s):
setup(s)
c = nda(s)
cid = c["contract_id"]
c = s.as_("owner").post(f"{C}/contracts/{cid}/sections/2/position", {"clause_id": "liability_fb"}).ok().body
seq = c["approvals"][-1]["seq"]
s.as_("legal").post(f"{C}/contracts/{cid}/approvals/{seq}/decide", {"decision": "approve"}).ok()
s.given("an approved draft with fingerprint F1")
f1 = show(s, cid).ok().body
s.check(f1["state"] == "APPROVED", "approved", f1["state"])
s.when("the owner changes one variable (F2) and tries to send")
f2 = s.as_("owner").post(f"{C}/contracts/{cid}/variables", {"values": {"notice_days": 45}}).ok().body
s.then("the approval is void, the draft is IN_APPROVAL for F2, and send is refused until F2 is approved")
s.check(f2["draft_fingerprint"] != f1["draft_fingerprint"] and f2["state"] == "IN_APPROVAL", "F2 in approval", f2)
s.check([a["state"] for a in f2["approvals"] if a["draft_fingerprint"] == f1["draft_fingerprint"]] == ["void"],
"F1's approval is void", f2["approvals"])
send(s, cid).refused(409)
seq2 = [a for a in f2["approvals"] if a["current"] and a["state"] == "pending"][0]["seq"]
s.as_("legal").post(f"{C}/contracts/{cid}/approvals/{seq2}/decide", {"decision": "approve"}).ok()
send(s, cid).ok()
@CAT.scenario("SC-CT-03", "a variable outside its band", stories=["CT-03", "CT-07", "CT-50"])
def outside_band(s):
setup(s)
s.given("a freelance template whose day_rate band is 300–600 GBP, and hr sending 750")
bp = person(s)
src = f"hr:onboarding:{s.key('R')}#3"
inp = {"type_code": "freelance_agreement", "counterparty_bp_id": int(bp), "context_ref": "p|1", "source_ref": src,
"variables": {"start_date": "2026-11-02", "end_date": "2027-03-31", "rate_amount": "750.0000",
"rate_basis": "day", "currency": "GBP"}, "signers": signers(s)}
s.when("contracts.draft runs")
r = cap(s, "hr_service", "contracts.draft", f"{src}|freelance_agreement", inp).ok().body
s.then("drafted with deviations [day rate above band] and approval_required true; send is refused")
out = r["output"]
s.check(r["status"] == "done" and out["approval_required"] is True and out["state"] == "IN_APPROVAL", "drafted", r)
s.check(any("above its band" in d["what"] and d["needs"] == "finance" for d in out["deviations"]), "the band", out)
sent = cap(s, "hr_service", "contracts.send", f"{src}|send", {"contract_id": out["contract_id"]}).ok().body
s.check(sent["status"] == "failed" and not sent["retryable"] and "approval by finance" in sent["message"],
"send refused in words", sent)
@CAT.scenario("SC-CT-04", "an approval nobody answers", stories=["CT-07"], tags=["clock"])
def nobody_answers(s):
setup(s)
s.clock.set("2026-10-19T10:00 Europe/London")
c = nda(s)
cid = c["contract_id"]
c = s.as_("owner").post(f"{C}/contracts/{cid}/sections/2/position", {"clause_id": "liability_fb"}).ok().body
seq = c["approvals"][-1]["seq"]
s.given("an approval with a 2-working-day timeout and a fallback group")
s.when("the timeout passes undecided")
s.clock.advance("+3 working days")
s.tick()
s.then("it is escalated to the fallback group; the original approver can no longer decide it")
s.as_("legal").post(f"{C}/contracts/{cid}/approvals/{seq}/decide", {"decision": "approve"}).not_found()
s.check(not [a for a in s.as_("legal").get(f"{C}/approvals").ok().body["approvals"] if a["contract_id"] == cid],
"gone from legal's queue")
q = s.as_("director").get(f"{C}/approvals").ok().body["approvals"]
s.check(any(a["contract_id"] == cid and a["state"] == "escalated" for a in q), "in the director's queue", q)
c = show(s, cid).ok().body
s.check(c["approvals"][-1]["escalated_at"] is not None, "the escalation is recorded", c["approvals"])
s.as_("director").post(f"{C}/contracts/{cid}/approvals/{seq}/decide", {"decision": "approve"}).ok().field(
"contract_state", "APPROVED")
# ── SC-CT-05 … 07: links and versions ──────────────────────────────────────────────────────
@CAT.scenario("SC-CT-05", "a signer looks at a stale version", stories=["CT-09", "CT-11"], tags=["public"])
def stale_version(s):
setup(s)
c = nda(s)
cid = c["contract_id"]
v1 = send(s, cid).ok().body
url1 = copy_link(s, cid)
s.given("version 1 out for signature; the starter opens the link")
s.link(url1).open().status(200)
s.when("the owner recalls and sends version 2, and the starter submits from the page still showing version 1")
s.as_("owner").post(f"{C}/contracts/{cid}/recall", {"reason": "wrong start date"}).ok()
s.as_("owner").post(f"{C}/contracts/{cid}/variables", {"values": {"start_date": "2026-11-08"}}).ok()
v2 = send(s, cid).ok().body
s.check(v2["version"] == 2 and v2["sha256"] != v1["sha256"], "version 2 frozen", v2)
r = sign_link(s, url1, v1["sha256"], 1)
s.then("refused; no SIGNED event for version 1; the version-1 link is revoked; the version-2 link works")
s.check(r.status_code == 404 and "no longer works" in r.text, "the version-1 link no longer works", r.text[:200])
r = sign_us(s, cid, v1["sha256"], 1)
s.check(r.status_code == 409 and "re-issued" in r.reason, "our side on version 1: refused in words", r.reason)
sign_link(s, copy_link(s, cid), v2["sha256"], 2).status(200)
ev = s.as_("contract_admin").get(f"{C}/contracts/{cid}/certificate").ok().body["events"]
s.check(not [e for e in ev if e["kind"] == "SIGNED" and e["version"] == 1], "no SIGNED on version 1", ev)
s.check([e["version"] for e in ev if e["kind"] == "SIGNED"] == [2], "one SIGNED, on version 2", ev)
@CAT.scenario("SC-CT-06", "signing out of order", stories=["CT-09"])
def out_of_order(s):
setup(s)
c = nda(s)
cid = c["contract_id"]
v = send(s, cid).ok().body
s.given("signing order starter (1) then director (2)")
s.when("the director signs first")
r = sign_us(s, cid, v["sha256"]).refused(409)
s.then("refused, naming who it waits on; after the starter signs, the director's signature succeeds")
s.check("waiting on Sam Nine" in r.reason, "names the starter", r.reason)
sign_link(s, copy_link(s, cid), v["sha256"]).status(200)
sign_us(s, cid, v["sha256"]).ok().field("state", "SIGNED")
@CAT.scenario("SC-CT-07", "a link expired, revoked or reused", stories=["CT-53", "CT-30"], tags=["public", "clock"])
def dead_links(s):
setup(s)
s.clock.set("2026-10-19T10:00 Europe/London")
a, b, c = nda(s), nda(s), nda(s)
for x in (a, b, c):
x["v"] = send(s, x["contract_id"]).ok().body
s.given("three signing links")
url_a = copy_link(s, a["contract_id"])
url_b_old = copy_link(s, b["contract_id"])
copy_link(s, b["contract_id"]) # reissued: the old one stops
url_c = copy_link(s, c["contract_id"])
sign_link(s, url_c, c["v"]["sha256"]).status(200)
s.when("one is used after its lifetime, one after a reissue, one after it already signed")
s.clock.advance("+15d")
s.tick()
for url in (url_a, url_b_old, url_c):
r = s.link(url).open()
s.check(r.status_code == 404 and "no longer works" in r.text, "this link no longer works", r.text[:200])
s.then("a contract past its lifetime unsigned is EXPIRED_UNSIGNED")
show(s, a["contract_id"]).field("state", "EXPIRED_UNSIGNED")
# ── SC-CT-08 … 10: elsewhere, amendments, key dates ────────────────────────────────────────
DEED_TEMPLATE = {"name": "Deed of assignment",
"slots": [{"slot_no": 1, "family": "ip_assignment", "clause_id": "ip_full", "clause_version": 1}],
"variables": [{"name": "counterparty_name", "type": "text", "source": "COUNTERPARTY"},
{"name": "start_date", "type": "date", "source": "DATES", "validation": {"term": "start"}},
{"name": "end_date", "type": "date", "source": "DATES", "validation": {"term": "end"}}]}
@CAT.scenario("SC-CT-08", "signed elsewhere", stories=["CT-12"])
def signed_elsewhere(s):
setup(s)
code = own_type(s, "deed", DEED_TEMPLATE, execution="DEED", allowed_bp_roles=["supplier"],
retention_rule="deed-12y")
bp, _ = org(s)
c = s.as_("owner").post(f"{C}/contracts", {"type_code": code, "bp_id": bp, "form_key": s.key("FK"),
"values": {"start_date": "2026-11-01", "end_date": "2027-10-31"},
"signers": signers(s)}).ok().body
cid = c["contract_id"]
s.given("a DEED type contract")
s.when("the owner tries to send it, then uploads the paper copy")
r = send(s, cid).refused(409)
s.check("not signed online" in r.reason, "deeds are not signed online here", r.reason)
import base64
import hashlib
paper = b"%PDF-1.4 scanned deed with wet signatures and a witness"
out = s.as_("owner").post(f"{C}/contracts/{cid}/signed-elsewhere", {
"filename": "deed.pdf", "content_base64": base64.b64encode(paper).decode(),
"signed_where": "at our London office, witnessed"}).ok().body
s.then("SIGNED_ELSEWHERE with the file's sha256; signers SIGNED_ELSEWHERE, never SIGNED; key dates exist")
s.check(out["stored_state"] == "SIGNED_ELSEWHERE" and out["executed"]["sha256"] == hashlib.sha256(paper).hexdigest()
and out["executed"]["provenance"] == "UPLOADED", "filed", out)
s.check({x["status"] for x in out["signers"]} == {"SIGNED_ELSEWHERE"}, "signers SIGNED_ELSEWHERE", out["signers"])
s.check({k["kind"] for k in out["key_dates"]} >= {"START", "END"}, "key dates", out["key_dates"])
ev = s.as_("contract_admin").get(f"{C}/contracts/{cid}/certificate").ok().body["events"]
s.check(not [e for e in ev if e["kind"] == "SIGNED"], "no SIGNED event", ev)
@CAT.scenario("SC-CT-09", "amendment and renewal", stories=["CT-14"])
def renewal(s):
setup(s)
bp, _ = org(s, roles=("supplier",))
body = {"type_code": "freelance_agreement", "bp_id": bp, "form_key": s.key("FK"), "signers": signers(s),
"values": {"start_date": "2026-11-02", "end_date": "2027-03-31", "rate_amount": "500.00",
"rate_basis": "day"}}
c = s.as_("owner").post(f"{C}/contracts", body).ok().body
cid = c["contract_id"]
v = send(s, cid).ok().body
sign_link(s, copy_link(s, cid), v["sha256"]).status(200)
sign_us(s, cid, v["sha256"]).ok()
s.given("a signed freelance agreement ending 2027-03-31")
s.when("a RENEWAL to 2027-06-30 is drafted and signed")
child = s.as_("owner").post(f"{C}/contracts/{cid}/renewal", {"ends_on": "2027-06-30"}).ok().body
kid = child["contract_id"]
s.check(child["amends_contract_id"] == cid and child["amendment_kind"] == "RENEWAL", "a renewal of it", child)
v2 = send(s, kid).ok().body
sign_link(s, copy_link(s, kid), v2["sha256"]).status(200)
sign_us(s, kid, v2["sha256"]).ok()
s.then("the original is SUPERSEDED, the renewal SIGNED, END and NOTICE_DEADLINE move, the original still verifies")
orig = show(s, cid).ok().body
ren = show(s, kid).ok().body
s.check(orig["stored_state"] == "SUPERSEDED" and ren["stored_state"] == "SIGNED", "states", (orig["stored_state"],
ren["stored_state"]))
s.check(all(k["done"] for k in orig["key_dates"]), "the original's dates are closed", orig["key_dates"])
s.check({(k["kind"], k["due_on"]) for k in ren["key_dates"]} >= {("END", "2027-06-30"), ("NOTICE_DEADLINE",
"2027-05-31")},
"the renewal's END and NOTICE_DEADLINE", ren["key_dates"])
s.as_("contract_admin").get(f"{C}/contracts/{cid}/evidence").ok().field("valid", True)
@CAT.scenario("SC-CT-10", "notice deadlines", stories=["CT-15", "CT-30"], tags=["clock"])
def notice_deadlines(s):
setup(s)
s.clock.set("2026-05-01T10:00 Europe/London")
s.given("a supplier contract ending 2026-12-31 with 90 days' notice and reminders at 120/90/60/30")
cid, _ = signed_nda(s, start_date="2026-05-01", end_date="2026-12-31", notice_days=90)
kd = {k["kind"]: k for k in show(s, cid).ok().body["key_dates"]}
s.check(kd["NOTICE_DEADLINE"]["due_on"] == "2026-10-02", "NOTICE_DEADLINE is 2 October", kd)
def reminders():
return [d for d in show(s, cid).ok().body["deliveries"] if d["kind"].startswith("REMIND:NOTICE_DEADLINE")]
s.when("the clock passes 2026-06-04, 2026-07-04, 2026-08-03 and 2026-09-02")
for i, day in enumerate(("2026-06-04", "2026-07-04", "2026-08-03", "2026-09-02"), start=1):
s.clock.set(f"{day}T09:00 Europe/London")
s.tick()
s.tick() # a second tick on a reminder date sends nothing more
s.check(len(reminders()) == i, f"one reminder by {day}", reminders())
s.then("one reminder at each of the four dates to the owner group")
s.check(sorted(int(d["kind"].rsplit(":", 1)[-1]) for d in reminders()) == [30, 60, 90, 120], "the slots",
reminders())
s.check({d["recipient"] for d in reminders()} == {"owner.1@scenario.test"}, "to the owner", reminders())
# ── SC-CT-11: tamper ───────────────────────────────────────────────────────────────────────
def _tamper(cid: str, seq: int) -> None:
"""Someone with bucket access edits one event's IP (in process: the brick's own storage)."""
from sqlalchemy import update
from brick_contracts.application import use_cases
from brick_contracts.infrastructure import tables as T
with use_cases.repository() as u:
u.conn.execute(update(T.signature_event).where(T.signature_event.c.contract_id == cid,
T.signature_event.c.seq == seq).values(ip="203.0.113.66"))
@CAT.scenario("SC-CT-11", "the hash chain tampered with", stories=["CT-13", "CT-32"], modes=["inprocess"])
def tampered(s):
setup(s)
cid, _ = signed_nda(s)
n = len(s.as_("contract_admin").get(f"{C}/contracts/{cid}/certificate").ok().body["events"])
s.given(f"a signed contract with {n} events")
s.when("someone with bucket access edits the IP on event 4")
_tamper(cid, 4)
s.then('the certificate reports "broken at seq 4"; verify_evidence fails; the executed copy is refused')
cert = s.as_("contract_admin").get(f"{C}/contracts/{cid}/certificate").ok().body
s.check(cert["chain"] == {"valid": False, "broken_at": 4} and "BROKEN at event 4" in cert["text"], "broken at 4",
cert["chain"])
s.as_("contract_admin").get(f"{C}/contracts/{cid}/evidence").ok().field("valid", False)
s.as_("contract_admin").get(f"{C}/contracts/{cid}/executed").refused(409)
s.note("and before the last signature: the executed copy is not produced")
c = nda(s)
v = send(s, c["contract_id"]).ok().body
sign_link(s, copy_link(s, c["contract_id"]), v["sha256"]).status(200)
_tamper(c["contract_id"], 2)
sign_us(s, c["contract_id"], v["sha256"]).ok().field("executed", False)
after = show(s, c["contract_id"]).ok().body
s.check(after["executed"] is None and after["note"].startswith("REFUSED"), "refused, and the owner told", after["note"])
# ── SC-CT-12 … 16 ──────────────────────────────────────────────────────────────────────────
@CAT.scenario("SC-CT-12", "no mail provider", stories=["CT-08", "CT-52"])
def no_mail(s):
setup(s)
s.given("a composition with no mail.send provider")
c = nda(s, sig=signers(s, extra_them="Kim Second"))
cid = c["contract_id"]
s.when("a contract is sent to two signers")
send(s, cid).ok()
t = [x for x in show(s, cid).ok().body["tasks"] if x["kind"] == "INVITE"]
s.then("two manual delivery tasks with the exact recipient and message")
s.check(len(t) == 2 and all(x["channel"] == "manual" and x["recipient"] and "copy the signing link" in x["message"]
for x in t), "two tasks", t)
s.check(all("/p/" not in x["message"] for x in t), "the link itself is never in a task", t)
url = copy_link(s, cid, t[0]["signer_no"])
s.check(url.startswith("http"), "copying shows the link once", url)
s.as_("owner").post(f"{C}/contracts/{cid}/deliveries/delivered",
{"signer_no": t[0]["signer_no"], "kind": "INVITE", "seq": t[0]["seq"]}).ok().field("status",
"done")
ev = s.as_("contract_admin").get(f"{C}/contracts/{cid}/certificate").ok().body["events"]
s.check([e["kind"] for e in ev if e["signer_no"] == t[0]["signer_no"]][-2:] == ["REISSUED", "DELIVERED"],
"REISSUED then DELIVERED (manual) in the chain", ev)
@CAT.scenario("SC-CT-13", "missing variables", stories=["CT-05", "CT-35", "CT-50"])
def missing_values(s):
setup(s)
no_holiday = {k: v for k, v in __import__("seed").ERA_DEFAULTS.items() if k != "holiday_entitlement"}
code = own_type(s, "employment", employment_template(defaults=no_holiday), allowed_bp_roles=["employee"],
allow_roleless_person=True, sensitivity="employment", pack="gb", pack_checks=["era_s1"])
bp = person(s)
src = f"hr:onboarding:{s.key('R')}#3"
inp = {"type_code": code, "counterparty_bp_id": int(bp), "context_ref": "p|1", "source_ref": src,
"variables": {"job_title": "Designer", "start_date": "2026-11-02", "rate_amount": "42000.0000",
"rate_basis": "year", "currency": "GBP"}, "signers": signers(s)}
s.given("an employment template needing the ERA s.1 particulars, and hr supplying no holiday entitlement")
s.when("contracts.draft runs")
r = cap(s, "hr_service", "contracts.draft", f"{src}|{code}", inp)
s.then("missing [holiday_entitlement] is returned and no partial contract exists")
s.check(r.status_code == 422 and "missing: holiday_entitlement" in r.reason, "missing, in words", r.reason)
from brick_contracts.domain.values import contract_id_for
show(s, contract_id_for(src, code)).not_found()
inp["variables"]["holiday_entitlement"] = "25 days plus bank holidays"
r = cap(s, "hr_service", "contracts.draft", f"{src}|{code}", inp).ok().body
s.check(r["status"] == "done" and r["output"]["missing"] == [], "a second call with it drafts", r)
@CAT.scenario("SC-CT-14", "IR35 refusal", stories=["CT-35"])
def ir35(s):
setup(s)
bp, _ = org(s, roles=("supplier",))
def draft(regime):
body = {"type_code": "freelance_agreement", "bp_id": bp, "form_key": s.key("FK"), "signers": signers(s),
"values": {"start_date": "2026-11-02", "end_date": "2027-03-31", "rate_amount": "500.00",
"engagement_via": "PSC", "intermediary_name": "Nine Services Ltd",
"client_size_regime": regime, "ir35_determination": "OUTSIDE"}}
return s.as_("owner").post(f"{C}/contracts", body).ok().body["contract_id"]
s.given("FREELANCE_SERVICES with PSC, MEDIUM_LARGE, OUTSIDE and no sds_issued_on")
cid = draft("medium")
s.when("the owner sends it")
r = send(s, cid).refused(409)
s.then("refused in words; with sds_issued_on and sds_reference it sends; with SMALL it sends without them")
s.check("status determination statement" in r.reason, "the IR35 words", r.reason)
s.as_("owner").post(f"{C}/contracts/{cid}/variables", {"values": {"sds_issued_on": "2026-10-15",
"sds_reference": "SDS-2026-14"}}).ok()
send(s, cid).ok()
send(s, draft("small")).ok()
@CAT.scenario("SC-CT-15", "principal statement due on day one", stories=["CT-35"], tags=["clock"])
def statement_due(s):
setup(s)
s.clock.set("2026-10-26T10:00 Europe/London")
bp = person(s)
src = f"hr:onboarding:{s.key('R')}#3"
inp = {"type_code": "employment_contract", "counterparty_bp_id": int(bp), "context_ref": "p|1", "source_ref": src,
"variables": {"job_title": "Designer", "start_date": "2026-11-02", "rate_amount": "42000.0000",
"rate_basis": "year"}, "signers": signers(s)}
cid = cap(s, "hr_service", "contracts.draft", f"{src}|employment_contract", inp).ok().body["output"]["contract_id"]
cap(s, "hr_service", "contracts.send", f"{src}|send", {"contract_id": cid}).ok()
s.given("an employment contract starting 2026-11-02, still OUT_FOR_SIGNATURE — and never given to the starter")
s.when("the tick runs that morning")
s.clock.set("2026-11-02T08:00 Europe/London")
s.tick()
st = cap(s, "hr_service", "contracts.status", f"status|{cid}", {"contract_id": cid}).ok().body["output"]
s.then('contracts.status warns hr ("the written statement has not been given and is due today")')
s.check(st["state"] == "OUT_FOR_SIGNATURE" and st["principal_statement_issued_at"] is None, "not given", st)
s.check(any("has not been given and is due today" in w for w in st["warnings"]), "the warning", st["warnings"])
s.check(any(k["kind"] == "PRINCIPAL_STATEMENT_DUE" and k["due_on"] == "2026-11-02" and not k["done"]
for k in st["key_dates"]), "the key date is due", st["key_dates"])
s.note("issued counts, not only signed: once it is delivered the warning goes")
t = [x for x in show(s, cid, ("hr_admin", 1)).ok().body["tasks"] if x["kind"] == "INVITE"][0]
s.as_("hr_admin").post(f"{C}/contracts/{cid}/deliveries/delivered", {"signer_no": t["signer_no"],
"kind": "INVITE", "seq": t["seq"]}).ok()
st = cap(s, "hr_service", "contracts.status", f"status|{cid}", {"contract_id": cid}).ok().body["output"]
s.check(st["principal_statement_issued_at"] and not st["warnings"], "given: no warning", st)
@CAT.scenario("SC-CT-16", "both sides must be named", stories=["CT-08", "CT-33"])
def both_sides(s):
setup(s)
c = nda(s, sig=[signers(s)[0]])
s.given("a draft with only the counterparty as signer")
s.when("the owner sends it")
r = send(s, c["contract_id"]).refused(409)
s.then('refused ("nobody on our side is named, so this cannot be complete"); readiness lists both_sides unmet')
s.check("nobody on our side is named, so this cannot be complete" in r.reason, "the words", r.reason)
rd = {x["key"]: x for x in show(s, c["contract_id"]).ok().body["readiness"]}
s.check(rd["both_sides"]["met"] is False, "both_sides unmet", rd)
# ── SC-CT-17 … 20 ──────────────────────────────────────────────────────────────────────────
@CAT.scenario("SC-CT-17", "decline and ask for changes", stories=["CT-10"], tags=["public"])
def decline_and_changes(s):
setup(s)
a, b = nda(s, sig=signers(s, extra_them="Kim Second")), nda(s, sig=signers(s, extra_them="Kim Second"))
va, vb = send(s, a["contract_id"]).ok().body, send(s, b["contract_id"]).ok().body
a_other = copy_link(s, a["contract_id"], 3)
b_other = copy_link(s, b["contract_id"], 3)
s.given("two contracts out for signature")
s.when("one signer declines with a reason and another asks for changes")
s.link(copy_link(s, a["contract_id"])).submit({"action": "decline", "reason": "the fee is wrong", "version": "1",
"sha256": va["sha256"]}).status(200)
s.link(copy_link(s, b["contract_id"])).submit({"action": "changes", "note": "please add a break clause",
"version": "1", "sha256": vb["sha256"]}).status(200)
s.then("one is DECLINED and one back in DRAFT; every other token on each is revoked; the owner is told")
ca, cb = show(s, a["contract_id"]).ok().body, show(s, b["contract_id"]).ok().body
s.check(ca["stored_state"] == "DECLINED" and cb["stored_state"] == "DRAFT", "states", (ca["stored_state"],
cb["stored_state"]))
for url in (a_other, b_other):
s.check(s.link(url).open().status_code == 404, "the other signer's link is revoked")
s.check(any(d["kind"] == "NOTICE:declined" for d in ca["deliveries"]) and
any(d["kind"] == "NOTICE:changes" for d in cb["deliveries"]), "the owner is told")
@CAT.scenario("SC-CT-18", "counterparty frozen at send", stories=["CT-51"])
def frozen_counterparty(s):
setup(s)
bp, legal = org(s)
c = nda(s, bp=bp)
v = send(s, c["contract_id"]).ok().body
s.given(f"a contract sent with the partner's legal name {legal!r}")
s.when("the partner is renamed in business-partner afterwards")
p = admin(s).get(f"/business-partner/partners/{bp}").ok().body
admin(s).put("/business-partner/partners", {"bp_key": p["bp_key"], "legal_name": "Renamed Holdings Ltd", "country": "GB"}).ok()
sign_link(s, copy_link(s, c["contract_id"]), v["sha256"]).status(200)
sign_us(s, c["contract_id"], v["sha256"]).ok()
s.then("the sent version and the executed copy still say the old name; a new draft shows the new name")
text = s.as_("contract_admin").get(f"{C}/contracts/{c['contract_id']}/versions").ok().body["versions"][0]["text"]
s.check(legal in text and "Renamed" not in text, "the frozen words", text[:300])
import base64
ex = base64.b64decode(s.as_("contract_admin").get(f"{C}/contracts/{c['contract_id']}/executed").ok().body[
"content_base64"])
s.check(legal.encode() in ex and b"Renamed" not in ex, "the executed copy")
c2 = nda(s, bp=bp)
s.check(any(v["value"] == "Renamed Holdings Ltd" for v in c2["variables"] if v["name"] == "counterparty_name"),
"a new draft has the new name", c2["variables"])
@CAT.scenario("SC-CT-19", "a counterparty of the wrong role", stories=["CT-02", "CT-50"])
def wrong_role(s):
setup(s)
bp, _ = org(s, roles=("supplier",))
s.given("an EMPLOYMENT type allowing only the employee role")
s.when("a draft is requested for a partner holding only the supplier role")
r = s.as_("hr_admin").post(f"{C}/contracts", {"type_code": "employment_contract", "bp_id": bp,
"form_key": s.key("FK"), "values": {}})
s.then('refused ("an employment contract needs a partner with the employee role")')
s.check(r.status_code in (409, 422) and "an employment contract needs a partner with the employee role" in r.reason,
"the words", r.reason)
src = f"hr:onboarding:{s.key('R')}#3"
r = cap(s, "hr_service", "contracts.draft", f"{src}|employment_contract", {
"type_code": "employment_contract", "counterparty_bp_id": int(bp), "context_ref": None, "source_ref": src,
"variables": {}, "signers": None})
s.check(r.status_code == 422, "through the capability: 422", r.reason)
def employment_signed(s, owner=("contract_admin", 1)):
bp = person(s)
body = {"type_code": "employment_contract", "bp_id": bp, "form_key": s.key("FK"), "signers": signers(s),
"values": {"job_title": "Designer", "start_date": "2026-11-02", "rate_amount": "42000.00",
"rate_basis": "year"}}
cid = s.as_(*owner).post(f"{C}/contracts", body).ok().body["contract_id"]
v = send(s, cid, owner).ok().body
sign_link(s, copy_link(s, cid, who=owner), v["sha256"]).status(200)
sign_us(s, cid, v["sha256"]).ok()
return cid
BODY_ROUTES = ("show_versions", "show_certificate", "verify_evidence", "fetch_executed")
@CAT.scenario("SC-CT-20", "employment bodies are sensitive-grade", stories=["CT-13"], tags=["permissions"])
def employment_bodies(s):
setup(s)
s.given("a signed employment contract")
cid = employment_signed(s)
own = s.as_("owner").post(f"{C}/contracts", {"type_code": "employment_contract", "bp_id": person(s),
"form_key": s.key("FK"), "values": {"job_title": "Designer"}}).ok().body
s.check(own["body"] == "withheld", "an owner without contract:employment:read drafts it and sees no body", own)
s.when("a manager, an owner without contract:employment:read, and HR open them through the API and MCP")
for who, c in (("manager", cid), ("owner", own["contract_id"])):
for door in ("http", "mcp"):
r = s.as_(who).call("show_contract", {"contract_id": c}, door=door).ok()
s.check(r.body["body"] == "withheld" and "sections" not in r.body and r.body["key_dates"] is not None,
f"{who} sees state and key dates only ({door})", r.body)
for uc in BODY_ROUTES:
if uc in ("show_certificate", "verify_evidence") and who == "manager":
continue # the matrix: a manager lacks contract:evidence:read (403)
s.as_(who).call(uc, {"contract_id": c}, door=door).not_found()
s.then("the first two get 404 on the body, versions, executed copy and certificate; HR sees everything")
for uc in BODY_ROUTES:
s.as_("hr_admin").call(uc, {"contract_id": cid}).ok()
s.check(show(s, cid, ("hr_admin", 1)).ok().body["body"] == "shown", "HR sees the body")
# ── SC-CT-21 … 25 ──────────────────────────────────────────────────────────────────────────
@CAT.scenario("SC-CT-21", "retention and erasure", stories=["CT-34"], tags=["clock"])
def retention(s):
setup(s)
s.clock.set("2020-09-01T10:00 Europe/London")
old, _ = signed_nda(s, start_date="2020-09-01", end_date="2020-10-15") # ended 6 years + 2 days ago
code = own_type(s, "deed", DEED_TEMPLATE, execution="DEED", allowed_bp_roles=["supplier"],
retention_rule="deed-12y")
import base64
bp, _ = org(s)
deed = s.as_("owner").post(f"{C}/contracts", {"type_code": code, "bp_id": bp, "form_key": s.key("FK"),
"values": {"start_date": "2019-01-01", "end_date": "2019-10-17"},
"signers": signers(s)}).ok().body["contract_id"]
s.as_("owner").post(f"{C}/contracts/{deed}/signed-elsewhere", {"filename": "deed.pdf", "signed_where": "London",
"content_base64": base64.b64encode(b"%PDF deed").decode()}).ok()
s.clock.set("2026-09-07T10:00 Europe/London")
dead = nda(s)["contract_id"]
s.as_("owner").post(f"{C}/contracts/{dead}/cancel", {"reason": "deal lost"}).ok()
live_sig = signers(s)
live = nda(s, sig=live_sig)["contract_id"]
send(s, live).ok()
s.given("a cancelled draft 40 days old (window 30), a simple contract ended over 6 years ago, a deed ended 7 years "
"ago, and an erasure request for a signer of a live contract")
s.clock.set("2026-10-17T10:00 Europe/London")
s.when("the sweep runs and the erasure is applied")
out = admin(s).post(f"{C}/retention/purge", {"erase_signer_email": live_sig[0]["email"]}).ok().body["purged"]
s.then("the draft is purged; the simple contract and its evidence are purged; the deed is kept; the live "
"contract's evidence is kept under the legal-claims exception")
gone = {x["contract_id"] for x in out if not x.get("kept")}
s.check({dead, old} <= gone and deed not in gone and live not in gone, "what went", out)
s.check(any(x["contract_id"] == live and x.get("kept") for x in out), "the live one kept, the reason recorded", out)
show(s, old).not_found()
show(s, deed).ok()
again = admin(s).post(f"{C}/retention/purge", {"erase_signer_email": live_sig[0]["email"]}).ok().body["purged"]
s.check(not [x for x in again if x["contract_id"] in (dead, old)], "running twice changes nothing", again)
@CAT.scenario("SC-CT-22", "double send and double sign", stories=["CT-08", "CT-09", "CT-50"],
tags=["concurrency", "public"])
def double_send_sign(s):
setup(s)
c = nda(s)
cid = c["contract_id"]
s.given("a ready draft")
s.when("send is clicked twice and a signer double-clicks sign")
rs = s.concurrently(2, lambda i: send(s, cid))
rs.expect({200: 2})
s.check(rs[0].body["sha256"] == rs[1].body["sha256"] and rs[0].body["version"] == 1, "one version", rs)
url = copy_link(s, cid)
sha = rs[0].body["sha256"]
two = s.concurrently(2, lambda i: sign_link(s, url, sha))
s.then("one version, one set of tokens, one SIGNED event per signer")
s.check(sorted(r.status_code for r in two) in ([200, 404], [200, 409]), "one signature", [r.status_code for r in two])
c = show(s, cid).ok().body
s.check(len(c["versions"]) == 1 and len([d for d in c["deliveries"] if d["kind"] == "INVITE"]) == 1,
"one version, one delivery", c)
ev = s.as_("contract_admin").get(f"{C}/contracts/{cid}/certificate").ok().body["events"]
s.check(len([e for e in ev if e["kind"] == "SIGNED" and e["signer_no"] == 1]) == 1, "one SIGNED", ev)
sign_us(s, cid, sha).ok()
sign_us(s, cid, sha).ok()
ev = s.as_("contract_admin").get(f"{C}/contracts/{cid}/certificate").ok().body["events"]
s.check(len([e for e in ev if e["kind"] == "SIGNED"]) == 2, "signing twice is one signature", ev)
@CAT.scenario("SC-CT-23", "the whole thing through the clock",
stories=["CT-05", "CT-07", "CT-08", "CT-09", "CT-31", "CT-30"], tags=["clock", "public"])
def through_the_clock(s):
setup(s)
s.clock.set("2026-11-02T09:00 Europe/London") # a Monday
bp = person(s, "Ana Freelancer")
src = f"hr:onboarding:{s.key('R')}#3"
s.given("hr's onboarding run asks for a freelance agreement with a rate above band, no mail provider")
inp = {"type_code": "freelance_agreement", "counterparty_bp_id": int(bp), "context_ref": "p|1", "source_ref": src,
"variables": {"start_date": "2026-11-16", "end_date": "2027-03-31", "rate_amount": "700.0000",
"rate_basis": "day", "currency": "GBP", "client_size_regime": "small"},
"signers": signers(s, them="Ana Freelancer")}
d = cap(s, "hr_service", "contracts.draft", f"{src}|freelance_agreement", inp).ok().body["output"]
cid = d["contract_id"]
s.check(d["approval_required"], "the rate needs approval", d)
s.when("draft, approval, send (day 0), the starter signs (day 1), a reminder to the director (day 4), the "
"director signs (day 5)")
seq = show(s, cid, ("hr_admin", 1)).ok().body["approvals"][-1]["seq"]
s.as_("finance").post(f"{C}/contracts/{cid}/approvals/{seq}/decide", {"decision": "approve"}).ok()
sent = cap(s, "hr_service", "contracts.send", f"{src}|send", {"contract_id": cid}).ok().body
s.check(sent["status"] == "done" and sent["output"]["state"] == "OUT_FOR_SIGNATURE", "sent", sent)
url = copy_link(s, cid, who=("hr_admin", 1))
s.clock.advance("+1 working days")
sign_link(s, url, sent["output"]["sha256"], name="Ana Freelancer").status(200)
s.clock.advance("+3 working days")
s.tick()
c = show(s, cid, ("hr_admin", 1)).ok().body
s.check([x["kind"] for x in c["deliveries"] if x["signer_no"] == 2] == ["CHASE:1"], "the director is reminded",
c["deliveries"])
s.clock.advance("+1 working days")
sign_us(s, cid, sent["output"]["sha256"]).ok().field("executed", True)
s.then("the director signs in the app (INTERNAL, POOL_SESSION); the executed copy exists once; hr fetches it")
ev = s.as_("hr_admin").get(f"{C}/contracts/{cid}/certificate").ok().body["events"]
mine = [e for e in ev if e["kind"] == "SIGNED" and e["signer_no"] == 2]
s.check(mine and mine[0]["signer_kind"] == "INTERNAL" and mine[0]["auth_method"] == "POOL_SESSION", "ours", mine)
st = cap(s, "hr_service", "contracts.status", f"status|{cid}", {"contract_id": cid}).ok().body["output"]
s.check(st["state"] in ("SIGNED", "ACTIVE") and st["executed"]["sha256"], "status reports it to hr", st)
s.tick()
f = cap(s, "hr_service", "contracts.fetch_executed", f"fetch|{cid}", {"contract_id": cid}).ok().body["output"]
import base64
import hashlib
s.check(hashlib.sha256(base64.b64decode(f["content_base64"])).hexdigest() == f["sha256"] == st["executed"]["sha256"],
"hr fetches the bytes", f["sha256"])
@CAT.scenario("SC-CT-24", "only the originator gets the executed copy", stories=["CT-50"], tags=["permissions"])
def originator_only(s):
setup(s)
bp = person(s)
src = f"hr:onboarding:{s.key('R')}#3"
inp = {"type_code": "employment_contract", "counterparty_bp_id": int(bp), "context_ref": "p|1", "source_ref": src,
"variables": {"job_title": "Designer", "start_date": "2026-11-02", "rate_amount": "42000.0000",
"rate_basis": "year"}, "signers": signers(s)}
cid = cap(s, "hr_service", "contracts.draft", f"{src}|employment_contract", inp).ok().body["output"]["contract_id"]
v = cap(s, "hr_service", "contracts.send", f"{src}|send", {"contract_id": cid}).ok().body["output"]
sign_link(s, copy_link(s, cid, who=("hr_admin", 1)), v["sha256"]).status(200)
sign_us(s, cid, v["sha256"]).ok()
s.given("an employment contract created by hr and signed")
s.when("hr, then crm's adapter identity, call contracts.status and contracts.fetch_executed")
hr = cap(s, "hr_service", "contracts.status", f"s|{cid}", {"contract_id": cid}).ok().body["output"]
f = cap(s, "hr_service", "contracts.fetch_executed", f"f|{cid}", {"contract_id": cid}).ok().body["output"]
crm = cap(s, "crm_service", "contracts.status", f"s|{cid}", {"contract_id": cid}).ok().body["output"]
s.then("hr receives the object key, sha256 and bytes; crm gets executed = null and 404 from fetch_executed")
s.check(hr["executed"] and hr["executed"]["sha256"] == f["sha256"], "hr", hr)
s.check(crm["executed"] is None, "crm: null", crm)
cap(s, "crm_service", "contracts.fetch_executed", f"f|{cid}", {"contract_id": cid}).not_found()
s.note("neither holds contract:employment:read")
crm_draft = cap(s, "crm_service", "contracts.draft", f"{src}x|employment_contract", {**inp, "source_ref": src + "x"})
s.check(crm_draft.status_code == 403, "crm cannot draft under hr:", crm_draft.reason)
@CAT.scenario("SC-CT-25", "our side signs in the app, not by link", stories=["CT-17", "CT-08"], tags=["public"])
def our_side_in_app(s):
setup(s)
c = nda(s)
cid = c["contract_id"]
s.given("a contract with an EXTERNAL signer (the starter) and an INTERNAL signer (a director)")
s.when("it is sent")
v = send(s, cid).ok().body
s.then("one public link is issued (the starter's); the director has none and signs from To sign")
s.check(v["deliveries"] == [{"signer_no": 1, "channel": "manual"}], "one link, one delivery", v)
sig = {x["signer_no"]: x for x in show(s, cid).ok().body["signers"]}
s.check(sig[1]["has_link"] and not sig[2]["has_link"], "the director has no link", sig)
s.as_("owner").post(f"{C}/contracts/{cid}/signers/2/copy-link", {}).refused(409)
s.as_("our_signatory").post(f"{C}/contracts/{cid}/sign", {"version": 1, "sha256": v["sha256"], "typed_name": "X",
"consent": True}).not_found()
sign_link(s, copy_link(s, cid), v["sha256"]).status(200)
s.check(cid in [x["contract_id"] for x in s.as_("director").get(f"{C}/to-sign").ok().body["to_sign"]],
"on the director's To sign list")
s.as_("director").post(f"{C}/contracts/{cid}/open-as-us", {}).ok()
sign_us(s, cid, v["sha256"]).ok()
ev = s.as_("contract_admin").get(f"{C}/contracts/{cid}/certificate").ok().body
signed = {e["signer_no"]: e for e in ev["events"] if e["kind"] == "SIGNED"}
s.check(signed[1]["auth_method"] == "EMAIL_LINK" and signed[2]["auth_method"] == "POOL_SESSION"
and signed[2]["pool_sub"] == "director.1", "EMAIL_LINK for the starter, POOL_SESSION + sub for us", signed)
s.check("POOL_SESSION" in ev["text"] and "pool sub director.1" in ev["text"], "on the certificate", ev["text"][:400])
# ── generated probes ───────────────────────────────────────────────────────────────────────
def _rows(s):
setup(s)
return {"contract_id": nda(s)["contract_id"]}
def _fresh_draft(s, who=("owner", 1)):
return nda(s, who=who)["contract_id"]
def _sent(s):
cid = _fresh_draft(s)
return cid, send(s, cid).ok().body
def _with_approval(s):
cid = _fresh_draft(s)
c = s.as_("owner").post(f"{C}/contracts/{cid}/sections/2/position", {"clause_id": "liability_fb"}).ok().body
return cid, c["approvals"][-1]["seq"]
def _rejected(s):
cid, seq = _with_approval(s)
s.as_("legal").post(f"{C}/contracts/{cid}/approvals/{seq}/decide", {"decision": "reject", "reason": "no"}).ok()
return cid
def _signed(s):
setup(s)
cid, _ = signed_nda(s, end_date="2027-10-31", notice_days=30)
return cid
def _show(cid):
return lambda s, r: show(s, cid)
def _r(fn):
"""A recipe: setup first, then fn(s) → (persona, args[, observe])."""
def recipe(s):
setup(s)
return fn(s)
return recipe
def _draft_args(s):
bp, _ = org(s)
return {"type_code": "nda", "bp_id": bp, "form_key": s.key("FK"), "values": {"start_date": "2026-11-01"},
"signers": signers(s)}
def _upload(s):
import base64
cid = _fresh_draft(s)
return "owner", {"contract_id": cid, "filename": "paper.pdf", "signed_where": "their office",
"content_base64": base64.b64encode(b"%PDF paper " + cid.encode()).decode()}, _show(cid)
def _ct(s):
return "contract_admin", {"type_code": f"probe_{uniq()}", "label": "probe type",
"allowed_bp_roles": ["customer"], "retention_rule": "simple-6y",
"approval_policy": "standard"}
def _clause(s):
return "contract_admin", {"clause_id": f"pc_{uniq()}", "family": "probe",
"position": "PREFERRED", "title": "Probe", "body": "Probe words."}
def _tmpl(s):
t = freelance_template(f"pt-{uniq()}")
return "contract_admin", t
def _approve_tmpl(s):
_, t = _tmpl(s)
v = admin(s).post(f"{C}/templates", t).ok().body["version"]
return "contract_admin#2", {"template_id": t["template_id"], "version": v}
def _retire_tmpl(s):
p, a = _approve_tmpl(s)
admin(s, 2).post(f"{C}/templates/{a['template_id']}/versions/{a['version']}/approve", {}).ok()
return "contract_admin", a
def _decide(s):
cid, seq = _with_approval(s)
return "legal", {"contract_id": cid, "seq": seq, "decision": "approve"}
def _sign_us(s):
cid, v = _sent(s)
sign_link(s, copy_link(s, cid), v["sha256"]).status(200)
return "director", {"contract_id": cid, "version": 1, "sha256": v["sha256"], "typed_name": "Dana Director",
"consent": True}
def _decline_us(s):
cid, v = _sent(s)
sign_link(s, copy_link(s, cid), v["sha256"]).status(200)
return "director", {"contract_id": cid, "version": 1, "sha256": v["sha256"], "reason": "not authorised"}
def _delivered(s):
cid, _ = _sent(s)
t = [x for x in show(s, cid).ok().body["tasks"] if x["kind"] == "INVITE"][0]
return "owner", {"contract_id": cid, "signer_no": t["signer_no"], "kind": t["kind"], "seq": t["seq"]}, _show(cid)
def _key_date(s):
cid = _signed(s)
k = show(s, cid).ok().body["key_dates"][0]
return "owner", {"contract_id": cid, "kind": k["kind"], "seq": k["seq"]}
RECIPES = {
"draft_contract": _r(lambda s: ("owner", _draft_args(s))),
"edit_draft": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid, "title": "Mutual NDA (probe)"}, _show(cid)))(
_fresh_draft(s))),
"set_variables": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid, "values": {"notice_days": 60}},
_show(cid)))(_fresh_draft(s))),
"choose_position": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid, "position": 2,
"clause_id": "liability_alt"}, _show(cid)))(_fresh_draft(s))),
"override_section": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid, "position": 2,
"body": "Liability is unlimited for fraud only."},
_show(cid)))(_fresh_draft(s))),
"add_signer": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid, "side": "THEM", "full_name": "Kim Extra",
"email": s.email("kim"), "signing_order": 1},
_show(cid)))(_fresh_draft(s))),
"edit_signer": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid, "signer_no": 1, "title": "Partner"},
_show(cid)))(_fresh_draft(s))),
"request_approval": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid}, _show(cid)))(_rejected(s))),
"send_for_signature": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid}, _show(cid)))(_fresh_draft(s))),
"recall": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid, "reason": "probe"}, _show(cid)))(_sent(s)[0])),
"mark_delivered_by_hand": _r(_delivered),
"sign_as_us": _r(_sign_us),
"decline_as_us": _r(_decline_us),
"keep_contract_type": _r(_ct),
"keep_clause": _r(_clause),
"amend_clause": _r(lambda s: (lambda a: ("contract_admin", {"clause_id": a, "body": "Amended probe words."}))(
admin(s).post(f"{C}/clauses", _clause(s)[1]).ok().body["clause_id"])),
"retire_clause": _r(lambda s: (lambda a: ("contract_admin", {"clause_id": a}))(
admin(s).post(f"{C}/clauses", _clause(s)[1]).ok().body["clause_id"])),
"keep_template": _r(_tmpl),
"approve_template": _r(_approve_tmpl),
"retire_template": _r(_retire_tmpl),
"keep_settings": _r(lambda s: ("contract_admin", {"kind": "setting", "rows": [{"key": "link_lifetime_days",
"value": "14"}]})),
"install_contract_pack": _r(lambda s: ("contract_admin", {"pack": "gb"})),
"upload_signed_elsewhere": _r(_upload),
"cancel": _r(lambda s: (lambda cid: ("owner", {"contract_id": cid, "reason": "probe"}, _show(cid)))(_fresh_draft(s))),
"complete_key_date": _key_date,
"purge_by_retention": _r(lambda s: ("contract_admin", {})),
# commands that make something new each time (mass assignment only)
"decide_approval": _r(_decide),
"copy_signing_link": _r(lambda s: ("owner", {"contract_id": _sent(s)[0], "signer_no": 1})),
"reissue_link": _r(lambda s: ("owner", {"contract_id": _sent(s)[0], "signer_no": 1})),
"open_as_us": _r(lambda s: (lambda cid, v: (sign_link(s, copy_link(s, cid), v["sha256"]),
("director", {"contract_id": cid}))[1])(*_sent(s))),
"amend_contract": lambda s: ("owner", {"contract_id": _signed(s), "changes": {"notice_days": 45}}),
"renew_contract": lambda s: ("owner", {"contract_id": _signed(s), "ends_on": "2028-10-31"}),
"terminate_contract": lambda s: ("owner", {"contract_id": _signed(s), "terminates_on": "2027-01-31"}),
"record_administrative_change": lambda s: ("owner", {"contract_id": _signed(s), "note": "internal ref typo",
"reference": "PO-123"}),
}
BP_WHY = "business-partner's own command: probed in its own catalogue (brick-business-partner)"
EXCEPTIONS = {**{n: BP_WHY for n in ("upsert_partner", "set_partner_status", "remove_role", "link_partners",
"unlink_partners", "promote_partner", "sync_person")},
"run_due": "the tick's answer reports what THIS tick did (counts and its time); that a second tick does "
"nothing more is SC-CT-10's and SC-CT-07's check"}
# A manager holds contract:contract:read_all but not contract:employment:read; the matrix's placeholder rows do not
# exist, so every read is a 404 for every role — the generated cells need no override.
CAT.permission_matrix(rows=Rows(make=_rows, owner="owner", other="owner#2"))
CAT.idempotency(recipes=RECIPES, exceptions=EXCEPTIONS, ignore=())
CAT.mass_assignment(recipes=RECIPES, exceptions=EXCEPTIONS)