#!/usr/bin/env python
"""hr's stories on a REAL DataK3 bucket (erp_platform.livecheck): what SQLite cannot show — DataK3's DDL, its types
(DATE, NUMERIC(19,4), BOOLEAN), ON CONFLICT, and an optimistic version decided at COMMIT.
BUCKET=erp-platform-test DODIL_ORG=dodil python scripts/live_check.py
1. migrate: the real erp-migrate (plan, apply, VERIFY) of src/brick_hr/schema.sql on a bucket with no hr tables;
2. the doors (HR.app() in this process, gateway-shaped identities) over a real engine on the bucket:
the gb pack and an org (its steps stored with their keys), one change_playbook and its BPMN, a hire, a slice change, a run tick, a manual task completed, a manager's 404 on
sensitive data, and two saves from one version at once — one 200, one 409;
3. every hr table dropped at the end.
No deploy: the app runs here, the data lives in the bucket. Your CLI login is the credential (never printed).
"""
import json
import os
import pathlib
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
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"))
from erp_platform import livecheck as lc # noqa: E402
TIMES: dict = {}
def timed(label):
class T:
def __enter__(self):
self.t = time.monotonic()
def __exit__(self, *a):
TIMES[label] = round(time.monotonic() - self.t, 1)
return T()
def migrate(tables) -> bool:
lc.drop_all(tables)
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
from erp_platform import migrate as m
from brick_hr.application.use_cases import VERSION
return m.migrate(ROOT / "src" / "brick_hr" / "schema.sql", {"hr": VERSION}) == 0
def main() -> int:
import anyio
import httpx
from erp_platform import work
from brick_hr.application import use_cases
from brick_hr.infrastructure.tables import HrRepository, md
from conftest import headers
from seed import hire_body, seed
tables = [t.name for t in md.sorted_tables]
with timed("migrate"):
lc.check("erp-migrate plans, applies and VERIFIES schema.sql on the bucket", migrate(tables))
eng = lc.engine()
@contextmanager
def repo():
with eng.begin() as conn:
yield HrRepository(conn)
use_cases.repository = repo
from brick_hr.app import app
def call(method, path, who, body=None):
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://hr", timeout=120) as c:
return await c.request(method, path, headers=who, json=body)
r = anyio.run(go)
try:
return r.status_code, r.json()
except ValueError:
return r.status_code, r.text
HR, FIN, MGR, TICK = headers("hr_admin"), headers("finance"), headers("manager", sub="live.mgr"), headers("clock")
today = date.today()
start = today + timedelta(days=21)
with timed("configure (gb pack, org, catalogues)"):
seed(lambda p, b: call("POST", p, HR, b))
s, pbs = call("GET", "/playbooks", HR)
lc.check("the gb pack's starter playbooks are stored", s == 200 and len(pbs["playbooks"]) >= 8, s)
keys = lc.rows("SELECT step_key FROM hr_playbook_step WHERE playbook_id = 'employee-onboarding' AND version = 1 "
"ORDER BY step_no")
lc.check("every step is stored with its stable key (0.4.0's step_key column)",
[r["step_key"] for r in keys][:3] == ["starter_details", "approve_rate", "send_contract"], keys)
with timed("hire"):
s, h = call("POST", "/hires", HR, hire_body("Live", "Check", start.isoformat(), rate_amount="42000.1234",
rate_basis="year", currency="GBP"))
lc.check("a hire: one person, engagement 1, an onboarding run", s == 200 and h["run_id"].startswith("onboarding:"), h)
pid = h["person_id"]
rows = lc.rows(f"SELECT CAST(person_id AS VARCHAR) AS pid, version FROM hr_person WHERE person_id = {pid}")
lc.check("hr_person on the bucket with its derived 56-bit id and version 1",
len(rows) == 1 and rows[0]["pid"] == pid and int(rows[0]["version"]) == 1, rows)
pay = lc.rows(f"SELECT CAST(rate_amount AS VARCHAR) AS r, valid_from, valid_to FROM hr_pay WHERE person_id = {pid}")
lc.check("the rate is stored exactly to four places (NUMERIC(19,4))", pay and pay[0]["r"].startswith("42000.1234"), pay)
s, again = call("POST", "/hires", HR, hire_body("Live", "Check", start.isoformat(), rate_amount="42000.1234",
rate_basis="year", currency="GBP"))
lc.check("the same hire again is the same person and run (idempotent)", s == 200 and again == h, again)
runs = lc.rows(f"SELECT run_id FROM hr_run WHERE person_id = {pid}")
lc.check("one run on the bucket", len(runs) == 1, runs)
with timed("slice change"):
s, v = call("GET", f"/people/{pid}", HR)
change = (start + timedelta(days=30)).isoformat()
s, out = call("POST", f"/people/{pid}/pay", HR, {"effective_date": change, "reason": "review",
"rate_amount": "45000.0000", "rate_basis": "year",
"expected_version": v["version"]})
pay = lc.rows(f"SELECT valid_from, valid_to, CAST(rate_amount AS VARCHAR) AS r FROM hr_pay WHERE person_id = {pid} "
"ORDER BY valid_from")
lc.check("a slice change delimits the slice before (valid_to = new valid_from − 1)",
s == 200 and len(pay) == 2 and str(pay[0]["valid_to"])[:10] == (start + timedelta(days=29)).isoformat()
and str(pay[1]["valid_from"])[:10] == change, pay)
with timed("tick"):
s, t = call("POST", "/run-due-steps", TICK, {})
lc.check("the tick acts on the run (intake and contract become manual tasks: no providers here)",
s == 200 and any(x.startswith(h["run_id"]) for x in t["acted"]), t)
s, r = call("GET", f"/runs/{h['run_id']}", HR)
st = {x["step_no"]: x for x in r["steps"]}
lc.check("the contract step is a manual task with its plan", st["3"]["manual"] and st["3"]["state"] == "waiting", st["3"])
s, done = call("POST", f"/runs/{h['run_id']}/steps/3/complete", HR, {"attachment": {
"filename": "signed.pdf", "content_base64": "JVBERi0xLjQgc2lnbmVk"}})
docs = lc.rows(f"SELECT doc_id, access_class, sha256 FROM hr_document WHERE person_id = {pid}")
lc.check("completing it files the signed copy on the person (pay class) and the step is done",
s == 200 and done["state"] == "done" and docs and docs[0]["access_class"] == "pay", (done, docs))
s, _ = call("POST", f"/people/{pid}/sensitive", FIN, {"kind": "bank", "value": {"bank_sort_code": "10-20-30"}})
lc.check("finance records bank details", s == 200, _)
lc.check("a manager asking for the same route gets 404, not 403", call("GET", f"/people/{pid}/sensitive", MGR)[0] == 404)
with timed("version race"):
s, v = call("GET", f"/people/{pid}", HR)
bodies = [{"effective_date": (start + timedelta(days=60 + i)).isoformat(), "reason": "race",
"rate_amount": f"4600{i}.0000", "rate_basis": "year", "expected_version": v["version"]} for i in range(2)]
with ThreadPoolExecutor(2) as pool:
res = list(pool.map(lambda b: call("POST", f"/people/{pid}/pay", HR, b), bodies))
statuses = sorted(x[0] for x in res)
lc.check("two saves from one version at once: one commits, the other 409 'changed since you opened it'",
statuses == [200, 409] and "changed since you opened it" in str([x[1] for x in res if x[0] == 409]), res)
pay = lc.rows(f"SELECT valid_from FROM hr_pay WHERE person_id = {pid} AND superseded_by IS NULL")
lc.check("the loser wrote nothing (three pay slices, not four)", len(pay) == 3, pay)
with timed("account step falls due"):
cal = date.fromisoformat(st["5"]["due_at"][:10]) if st["5"]["due_at"] else start
work.set_clock(at=work.parse_instant(f"{cal.isoformat()}T09:05 Europe/London"), freeze=True)
call("POST", f"/people/{pid}/compliance", HR, {"kind": "right_to_work_check", "checked_on": today.isoformat(),
"checked_by": "hr", "check_type": "manual_document_check"})
s, t = call("POST", "/run-due-steps", TICK, {})
s, mt = call("GET", "/my-tasks", headers("it"))
work.reset_clock()
acc = [x for x in mt.get("overdue", []) + mt.get("today", []) if x["run_id"] == h["run_id"] and x["step_no"] == "5"]
lc.check("the account step is a manual task for IT carrying the exact input",
acc and acc[0]["capability"] == "identity.provision_user"
and acc[0]["input"]["primary_email"] == "live.check@co.example", acc or mt)
with timed("change_playbook"):
s, ch = call("POST", "/playbooks/employee-onboarding/changes", HR, {
"change": "add_step", "before": ["send_contract"],
"step": {"kind": "approval", "title": "Director approves the contract", "responsible_group": "director"}})
lc.check("one change by key is a new version, and its deviations say so",
s == 200 and ch["version"] == 2 and ch["key"] == "director_approves_the_contract"
and ch["summary"]["foundation"] == 0, ch)
s, x = call("GET", "/playbooks/employee-onboarding/bpmn", HR)
lc.check("the BPMN of v2 carries the keyed ids and marks the deviation",
s == 200 and 'id="Task_director_approves_the_contract"' in str(x) and 'dodil:deviation="variation"' in str(x))
with eng.begin() as conn:
lc.check("nothing to backfill: every stored step has its key", HrRepository(conn).playbooks.backfill_keys() == 0)
print("\ntimings (s):", json.dumps(TIMES))
eng.dispose()
return lc.finish(tables)
if __name__ == "__main__":
raise SystemExit(main())