#!/usr/bin/env python
"""contracts' stories on a REAL DataK3 bucket: what SQLite cannot show — DataK3's DDL and types, ON CONFLICT, the
optimistic version and the single-use link decided at COMMIT, the signing transaction sharing one engine with the
platform's link tables (tier 1: one bucket), and how long each step takes.
BUCKET=erp-platform-test DODIL_ORG=dodil python scripts/live_check.py
1. migrate: the real erp-migrate (plan, apply, VERIFY) of src/brick_contracts/schema.sql, business-partner's
schema.sql and the platform's own tables, on a bucket without them — the tables are built from the schema files,
never from the in-code definitions, so a NOT NULL or type difference shows up here;
2. the composition (contracts + the REAL business-partner, tier 1) and its PUBLIC app, served in this process over
one real engine on the bucket, with gateway-shaped identities: the gb pack and a library; a draft with a Fallback
clause, legal's approval, send (no mail provider: a manual task), the signing page through the public app, the
counterparty's signature, our director's signature in the app, the executed copy (its bytes start with the signed
version's, both sha256 checked), the certificate; a double-click on sign → one signature; hr's identity drafting
through contracts.draft and crm's identity refused the executed copy; the tick; someone editing event 4's IP →
"broken at 4" and the executed copy refused;
3. every table this run created is dropped at the end. No deploy. Your CLI login is the credential (never printed).
"""
import base64
import hashlib
import json
import os
import pathlib
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import ExitStack
from datetime import date, timedelta
from urllib.parse import unquote, urlparse
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
sys.path.insert(0, str(ROOT / "tests"))
os.environ.setdefault("PATH", "")
os.environ["PATH"] = os.path.expanduser("~/.local/bin") + os.pathsep + os.environ["PATH"]
from erp_platform import livecheck as lc # noqa: E402
TIMES: dict = {}
STATEMENTS = {"n": 0}
class timed:
"""Seconds AND the number of SQL statements the step sent to the bucket (round trips are what DataK3 costs)."""
def __init__(self, label):
self.label = label
def __enter__(self):
self.t, self.n = time.monotonic(), STATEMENTS["n"]
def __exit__(self, *a):
TIMES[self.label] = {"s": round(time.monotonic() - self.t, 2), "statements": STATEMENTS["n"] - self.n}
def token() -> None:
cmd = [lc.DODIL, "data", "connect", lc.BUCKET, "-o", "json"] + (["--org", lc.ORG] if lc.ORG else [])
u = urlparse(json.loads(subprocess.run(cmd, capture_output=True, text=True, check=True, env=lc._cli_env()).stdout)["pg"])
os.environ["DODIL_TOKEN"] = unquote(u.password or "") # this process only; never printed
def exists(t: str) -> bool:
try:
lc.pg(f"SELECT 1 FROM {t} LIMIT 1", attempts=1)
return True
except RuntimeError:
return False
def main() -> int:
if not lc.BUCKET:
raise SystemExit("set BUCKET=erp-platform-test DODIL_ORG=dodil")
import importlib.resources
import yaml
from sqlalchemy import MetaData
from erp_platform import capabilities, public
from brick_contracts.infrastructure.tables import md
ours = [t.name for t in md.sorted_tables] + [capabilities.tables("contracts").claims.name,
capabilities.tables("contracts").log.name]
import re
def names(sql_path):
return re.findall(r"CREATE TABLE IF NOT EXISTS (\w+)", pathlib.Path(sql_path).read_text())
bp_sql = pathlib.Path(str(importlib.resources.files("brick_bp").joinpath("schema.sql")))
pf_sql = pathlib.Path(str(importlib.resources.files("erp_platform").joinpath("platform.sql")))
others = names(bp_sql) + names(pf_sql)
if exists("ct_contract"):
lc.drop_all(ours) # a previous run's leftovers
created = ours + [t for t in others if not exists(t)] # drop only what this run made
token()
os.environ.setdefault("PUBLIC_LINK_BASE_URL", "https://sign.check.example")
from erp_platform import migrate as m
from brick_contracts.application.use_cases import VERSION
with timed("migrate (contracts + bp + platform)"):
ok = m.migrate(ROOT / "src" / "brick_contracts" / "schema.sql", {"contracts": VERSION}) == 0
ok = ok and m.migrate(bp_sql, {"bp": "0.3.0"}) == 0
ok = ok and m.migrate_platform() == 0
lc.check("erp-migrate plans, applies and VERIFIES contracts', business-partner's and the platform's schema", ok)
import anyio
import httpx
from erp_platform import db, seam
from erp_platform.scenarios import Catalogue
from erp_platform.scenarios.inprocess import stage
from seed import seed
roles = yaml.safe_load((ROOT / "tests" / "scenarios" / "roles.yaml").read_text())["roles"]
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]})
ADMIN, ADMIN2, OWNER, LEGAL, DIRECTOR, CLOCK = (who("contract_admin"), who("contract_admin", 2), who("owner"),
who("legal"), who("director"), who("clock"))
HR_SVC = seam.gateway_headers({"sub": "hr-system", "email": "hr-system", "app_roles": ["service"],
"permissions": ["contract:capability:invoke", "bp:person:sync", "bp:person:read"]})
CRM_SVC = seam.gateway_headers({"sub": "crm-adapter", "email": "crm-adapter", "app_roles": ["service"],
"permissions": ["contract:capability:invoke"]})
from sqlalchemy import event as sa_event
@sa_event.listens_for(db.engine, "before_cursor_execute")
def _count(*_a):
STATEMENTS["n"] += 1
st = ExitStack()
cat = Catalogue("live", compose=ROOT / "tests" / "scenarios" / "compose.yaml",
roles=ROOT / "tests" / "scenarios" / "roles.yaml")
staged = stage(cat, st, lambda name: db.engine, create=False) # one bucket: every brick and the links
app, door, public_base = staged.app, staged.public_app, staged.public_base
def call(method, path, h=None, body=None, form=None, target=None, raw=False):
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=target or app),
base_url=public_base if target else "http://staff", timeout=180) as c:
if form is not None:
return await c.request(method, path, headers=h or {}, data=form)
return await c.request(method, path, headers=h or {}, json=body)
r = anyio.run(go)
if raw:
return r
try:
return r.status_code, r.json()
except ValueError:
return r.status_code, r.text
try:
with timed("configure: gb pack, 10 clauses, 3 templates approved"):
seed(lambda p, b: call("POST", p, ADMIN, b), lambda p, b: call("POST", p, ADMIN2, b), prefix="/contracts")
s, p = call("PUT", "/business-partner/partners", ADMIN, {"bp_key": f"live-ct-{os.getpid()}", "legal_name": "Studio Nine Ltd",
"country": "GB", "roles": ["supplier"]})
lc.check("a supplier partner in business-partner", s == 200, p)
bp = p["bp_id"]
start = (date.today() + timedelta(days=14)).isoformat()
sig = [{"side": "THEM", "full_name": "Sam Nine", "email": "sam@nine.check.example", "signing_order": 1},
{"side": "US", "full_name": "Dana Director", "email": "director.1@scenario.test", "title": "Director",
"signing_order": 2}]
with timed("draft (reads the partner through bp)"):
s, c = call("POST", "/contracts/contracts", OWNER, {"type_code": "nda", "bp_id": bp, "form_key": "live-1",
"values": {"start_date": start}, "signers": sig})
lc.check("a DRAFT with sections copied, the counterparty from bp, nothing missing",
s == 200 and c["state"] == "DRAFT" and not c["missing"] and
{v["name"]: v["value"] for v in c["variables"]}["counterparty_name"] == "Studio Nine Ltd", c)
cid = c["contract_id"]
rows = lc.rows(f"SELECT state, version, draft_fingerprint FROM ct_contract WHERE contract_id = '{cid}'")
lc.check("ct_contract on the bucket at optimistic version 1 with its fingerprint",
rows and int(rows[0]["version"]) == 1 and rows[0]["draft_fingerprint"] == c["draft_fingerprint"], rows)
s, again = call("POST", "/contracts/contracts", OWNER, {"type_code": "nda", "bp_id": bp, "form_key": "live-1",
"values": {"start_date": start}, "signers": sig})
lc.check("the same form key again is the same contract (no second row)",
again["contract_id"] == cid and len(lc.rows(f"SELECT contract_id FROM ct_contract WHERE contract_id = '{cid}'")) == 1)
with timed("choose the Fallback position (an edit + review)"):
s, c = call("POST", f"/contracts/contracts/{cid}/sections/2/position", OWNER, {"clause_id": "liability_fb"})
lc.check("a Fallback clause puts it IN_APPROVAL for legal", s == 200 and c["state"] == "IN_APPROVAL", c)
s, r = call("POST", f"/contracts/contracts/{cid}/send", OWNER, {})
lc.check("send is refused in words while the approval is open",
s == 409 and "approval by legal needed" in r["detail"], r)
with timed("approve (legal)"):
s, a = call("POST", f"/contracts/contracts/{cid}/approvals/{c['approvals'][-1]['seq']}/decide", LEGAL,
{"decision": "approve"})
lc.check("legal approves the fingerprint", s == 200 and a["contract_state"] == "APPROVED", a)
with timed("send: render, freeze, issue the link in the same transaction"):
s, sent = call("POST", f"/contracts/contracts/{cid}/send", OWNER, {})
lc.check("version 1 frozen with its sha256; one manual delivery (no mail provider)",
s == 200 and sent["version"] == 1 and sent["deliveries"] == [{"signer_no": 1, "channel": "manual"}], sent)
links = lc.rows(f"SELECT link_id, grant_type, token_hash FROM erp_public_link WHERE subject_id = '{cid}|1'")
lc.check("the platform holds ONE SIGN link for the external signer, as a hash",
len(links) == 1 and links[0]["grant_type"] == "SIGN" and len(links[0]["token_hash"]) == 64, links)
s, cl = call("POST", f"/contracts/contracts/{cid}/signers/1/copy-link", OWNER, {})
url = cl["url"]
path = "/" + url.split("://", 1)[1].split("/", 1)[1]
with timed("signing page (GET through the public app: OPENED)"):
s, page = call("GET", path, {"user-agent": "LiveCheck/1.0", "x-forwarded-for": "198.51.100.7"}, target=door)
lc.check("the signing page shows the words, the fingerprint and the consent", s == 200 and sent["sha256"] in page
and "I agree to sign" in page, str(page)[:200])
with timed("external signature (POST through the public app)"):
s, page = call("POST", path, {"user-agent": "LiveCheck/1.0", "x-forwarded-for": "198.51.100.7"},
form={"action": "sign", "typed_name": "Sam Nine", "consent": "yes", "version": "1",
"sha256": sent["sha256"]}, target=door)
lc.check("the counterparty signs electronically", s == 200 and "signed electronically" in page, str(page)[:200])
now_links = [r["link_id"] for r in lc.rows(
f"SELECT link_id FROM erp_public_link WHERE subject_id = '{cid}|1' AND grant_type = 'SIGN'")]
used = [r for lid in now_links for r in lc.rows(
f"SELECT kind FROM erp_public_link_event WHERE link_id = '{lid}' AND kind = 'used'")]
lc.check("the link's single-use claim committed with the signature", len(used) == 1, used)
s, again = call("GET", path, target=door)
lc.check("the spent link is the one dead-link page", s == 404, str(again)[:100])
with timed("our director signs in the app + the executed copy"):
s, out = call("POST", f"/contracts/contracts/{cid}/sign", DIRECTOR, {"version": 1, "sha256": sent["sha256"],
"typed_name": "Dana Director",
"consent": True})
lc.check("SIGNED, and the executed copy produced in the same request",
s == 200 and out["state"] == "SIGNED" and out["executed"] is True, out)
s, ex = call("GET", f"/contracts/contracts/{cid}/executed", ADMIN)
s2, v1 = call("GET", f"/contracts/contracts/{cid}/versions/1/file", ADMIN)
exb, v1b = base64.b64decode(ex["content_base64"]), base64.b64decode(v1["content_base64"])
lc.check("the executed copy is a real PDF whose sha256 matches, starting with the signed version's exact bytes",
exb[:5] == b"%PDF-" and hashlib.sha256(exb).hexdigest() == ex["sha256"] and exb.startswith(v1b)
and hashlib.sha256(v1b).hexdigest() == sent["sha256"], (ex.get("sha256"), len(exb), len(v1b)))
with timed("certificate (chain recomputed)"):
s, cert = call("GET", f"/contracts/contracts/{cid}/certificate", ADMIN)
kinds = [e["kind"] for e in cert["events"]]
lc.check("the certificate verifies: INVITED, OPENED, CONSENTED, SIGNED ×2, with IP and agent as fields",
cert["chain"]["valid"] and kinds.count("SIGNED") == 2 and "OPENED" in kinds and
all(e["ip"] == "198.51.100.7" for e in cert["events"] if e["signer_no"] == 1 and e["kind"] in
("OPENED", "CONSENTED", "SIGNED")), cert["chain"])
n_events = len(cert["events"])
stored = lc.rows(f"SELECT seq, row_hash FROM ct_signature_event WHERE contract_id = '{cid}' ORDER BY seq")
lc.check(f"{n_events} append-only events on the bucket, each with its row hash",
len(stored) == n_events and all(len(r["row_hash"]) == 64 for r in stored), stored)
# a double click on sign: DataK3 decides the single-use claim (and the version claim) at COMMIT
s, c2 = call("POST", "/contracts/contracts", OWNER, {"type_code": "nda", "bp_id": bp, "form_key": "live-2",
"values": {"start_date": start}, "signers": sig})
s, sent2 = call("POST", f"/contracts/contracts/{c2['contract_id']}/send", OWNER, {})
url2 = call("POST", f"/contracts/contracts/{c2['contract_id']}/signers/1/copy-link", OWNER, {})[1]["url"]
path2 = "/" + url2.split("://", 1)[1].split("/", 1)[1]
form = {"action": "sign", "typed_name": "Sam Nine", "consent": "yes", "version": "1", "sha256": sent2["sha256"]}
with timed("two signatures at once on one link"):
with ThreadPoolExecutor(2) as pool:
res = list(pool.map(lambda _: call("POST", path2, form=form, target=door), range(2)))
ev = lc.rows(f"SELECT seq FROM ct_signature_event WHERE contract_id = '{c2['contract_id']}' AND kind = 'SIGNED'")
lc.check("a double click is one signature (one 200, one refused) and ONE SIGNED event",
sorted(r[0] for r in res)[0] == 200 and sorted(r[0] for r in res)[1] in (404, 409) and len(ev) == 1,
([r[0] for r in res], ev))
# hr's identity through the capability; crm's identity never gets hr's copy
s, pp = call("POST", "/business-partner/partners/persons/sync", HR_SVC, {"person": {"person_id": str(os.getpid()),
"legal_name": "Anna Berg", "roles": []}})
src = f"hr:onboarding:live:{os.getpid()}#3"
inp = {"type_code": "freelance_agreement", "counterparty_bp_id": int(pp["bp_id"]), "context_ref": "p|1",
"source_ref": src, "variables": {"start_date": start, "end_date": (date.today() + timedelta(days=120)).isoformat(),
"rate_amount": "500.0000", "rate_basis": "day", "currency": "GBP",
"client_size_regime": "small"}, "signers": sig}
with timed("contracts.draft (hr's identity, through the capability claim)"):
s, d = call("POST", "/contracts/capabilities/contracts.draft", HR_SVC,
{"idempotency_key": f"{src}|freelance_agreement", "input": inp})
lc.check("hr drafts through contracts.draft: done, within band, no approval",
s == 200 and d["status"] == "done" and d["output"]["approval_required"] is False, d)
hcid = d["output"]["contract_id"]
s, d2 = call("POST", "/contracts/capabilities/contracts.draft", HR_SVC,
{"idempotency_key": f"{src}|freelance_agreement", "input": inp})
lc.check("the same call again is the stored answer (idempotent_replay), one contract",
d2.get("idempotent_replay") is True and d2["output"]["contract_id"] == hcid, d2)
s, crm = call("POST", "/contracts/capabilities/contracts.fetch_executed", CRM_SVC,
{"idempotency_key": f"f|{hcid}", "input": {"contract_id": hcid}})
lc.check("crm's identity asking for hr's contract's copy: 404", s == 404, crm)
with timed("tick (run_due): only what is due"):
s, t = call("POST", "/contracts/run-due", CLOCK, {})
lc.check("the tick runs", s == 200, t)
# someone with bucket access edits event 4's IP
lc.pg(f"UPDATE ct_signature_event SET ip = '203.0.113.66' WHERE contract_id = '{cid}' AND seq = 4")
s, cert = call("GET", f"/contracts/contracts/{cid}/certificate", ADMIN)
lc.check("the edited row is found: the chain is broken at seq 4", cert["chain"] == {"valid": False, "broken_at": 4},
cert["chain"])
s, r = call("GET", f"/contracts/contracts/{cid}/executed", ADMIN)
lc.check("and the executed copy is refused", s == 409 and "broken at event 4" in r["detail"], r)
finally:
st.close()
db.engine.dispose()
print("\ntimings (s):", json.dumps(TIMES, indent=1))
return lc.finish(created)
if __name__ == "__main__":
raise SystemExit(main())