#!/usr/bin/env python
"""The activity stories, end to end, on a real DataK3 bucket (erp_platform.livecheck).
BUCKET=<scratch bucket> DODIL_ORG=<org> python scripts/live_check.py
0.2.0 adds: six concurrent adds of one email on DataK3 give one contact and no error (the audit's 500), and the
optimistic version — a stale one is 409, and two edits racing from one version give one success.
"""
import pathlib
import sys
import threading
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from erp_platform import livecheck as lc # noqa: E402
from brick_activity.application import use_cases as uc # noqa: E402
from brick_activity.infrastructure.tables import ActivityRepository, EmailTaken, md # noqa: E402
ACME = 54848654143250435
def person(email, roles=("sales",)):
return {"sub": email, "email": email, "roles": list(roles), "auth_source": "pool",
"permissions": ["activity:contact:write", "activity:activity:write"]}
TOM, SOFIA = person("tom@co.example"), person("sofia@co.example")
def refused(fn, *a, **kw):
"""(status, reason) if refused — a Refused, or a lost version (what the doors answer 409) — else None."""
from erp_platform.repository import as_conflict
try:
return lc.refused(fn, *a, **kw)
except Exception as e: # noqa: BLE001 — re-raised unless a conflict
c = as_conflict(e)
if c is None:
raise
return 409, c.reason
def at_once(n, fn):
out = []
def go(i):
try:
out.append(("ok", fn(i)))
except Exception as e: # noqa: BLE001 — counted
r = None
try:
from erp_platform.repository import as_conflict
r = (409,) if as_conflict(e) else None
except Exception: # noqa: BLE001
pass
out.append(("refused", getattr(e, "status", None) or (r and r[0]) or type(e).__name__))
ts = [threading.Thread(target=go, args=(i,)) for i in range(n)]
for t in ts:
t.start()
for t in ts:
t.join()
return out
def main() -> int:
tables = [t.name for t in md.sorted_tables]
lc.create_schema(ROOT / "src" / "brick_activity" / "schema.sql", tables)
uc.repository = lc.repository(ActivityRepository)
anna = uc.add_contact(TOM, ACME, "Anna Berg", "Anna@Acme.com", title="Head of Marketing")
again = uc.add_contact(SOFIA, ACME, "Anna Berg", "anna@acme.com")
lc.check("one email, one contact — and the bp_id above 2^53 is exact",
again["contact_id"] == anna["contact_id"] and int(again["bp_id"]) == ACME, again)
try:
with uc.repository() as r:
r.claim_email("anna@acme.com", "C-INTRUDER")
took = None
except EmailTaken as e:
took = str(e)
owner = lc.rows("SELECT contact_id FROM activity_contact_emails WHERE email = 'anna@acme.com'")
lc.check("a second plain INSERT of a claimed email fails on DataK3 (23505), the claim unchanged",
took and owner == [{"contact_id": anna["contact_id"]}], (took, owner))
bo = uc.add_contact(TOM, ACME, "Bo Ek", "bo@acme.com")
taken = lc.refused(uc.edit_contact, TOM, bo["contact_id"], email="anna@acme.com")
uc.edit_contact(TOM, bo["contact_id"], email="bo.ek@acme.com")
claims = sorted(r["email"] for r in lc.rows("SELECT email FROM activity_contact_emails"))
lc.check("an email change cannot take someone else's, and frees the old address",
taken and taken[0] == 409 and claims == ["anna@acme.com", "bo.ek@acme.com"], (taken, claims))
uc.log_activity(TOM, "CALL", "first call", contact_id=anna["contact_id"], occurred_at="2026-10-01T09:00:00Z")
uc.log_activity(TOM, "CALL", "Rang Anna", contact_id=anna["contact_id"], note="proposal by Friday")
uc.log_activity(TOM, "NOTE", "about Acme", bp_id=ACME, occurred_at="2026-10-05T09:00:00Z")
tl = [a["subject"] for a in uc.timeline(TOM, bp_id=ACME)["activities"]]
lc.check("CRM-02 on the company, newest first, contacts' activities included",
tl == ["Rang Anna", "about Acme", "first call"], tl)
for subject, due in (("late", "2026-10-10"), ("now", "2026-10-16"), ("soon", "2026-10-20")):
uc.log_activity(TOM, "TASK", subject, bp_id=ACME, due_on=due)
board = uc.my_tasks(TOM, "2026-10-16")
lc.check("CRM-12 overdue / today / this week",
board["counts"] == {"overdue": 1, "today": 1, "this_week": 1}, board["counts"])
late = board["overdue"][0]["activity_id"]
uc.complete_task(TOM, late, "done at last")
stored = lc.rows(f"SELECT status, note, kind FROM activity_log WHERE activity_id = '{late}'")[0]
lc.check("completing keeps the task's record and adds the outcome",
stored == {"status": "done", "note": "done at last", "kind": "TASK"}, stored)
lc.check("a done task leaves the board", uc.my_tasks(TOM, "2026-10-16")["counts"]["overdue"] == 0)
# ── 0.2.0 ───────────────────────────────────────────────────────────────────────────────
out = at_once(6, lambda i: uc.add_contact(TOM, ACME, f"Race {i}", "race@acme.com")["contact_id"])
ids = {v for k, v in out if k == "ok"}
stored = lc.rows("SELECT contact_id FROM activity_contact_emails WHERE email = 'race@acme.com'")
lc.check("H12 six concurrent adds of one email on DataK3: one contact, every caller answered it (no 500)",
len(ids) == 1 and all(k == "ok" for k, _ in out) and stored == [{"contact_id": next(iter(ids))}],
(out, stored))
c = uc.get_contact(TOM, anna["contact_id"])
uc.edit_contact(TOM, anna["contact_id"], title="CMO", version=c["version"])
stale = refused(uc.edit_contact, TOM, anna["contact_id"], title="CEO", version=c["version"])
lc.check("C3 a stale version is 409 on DataK3 and writes nothing",
stale and stale[0] == 409 and uc.get_contact(TOM, anna["contact_id"])["title"] == "CMO", stale)
v = uc.get_contact(TOM, anna["contact_id"])["version"]
out = at_once(4, lambda i: uc.edit_contact(TOM, anna["contact_id"], phone=f"+46 {i}", version=v))
lc.check("C3 four edits racing from one version: one success, the rest 409",
sum(k == "ok" for k, _ in out) == 1 and all(x == ("refused", 409) for x in out if x[0] != "ok"), out)
return lc.finish(tables)
if __name__ == "__main__":
raise SystemExit(main())