#!/usr/bin/env python
"""The partner 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: the ADDITIVE migration from a 0.1 bucket (a business_partner without the person columns, with an
organisation in it) through the real `erp-migrate` code, and the person-partner stories BP-05..BP-07.
0.3.0 adds: the optimistic version (the column and its claim table, migrated from that 0.1 bucket too): a stale
version is 409 on DataK3, and two stewards racing from one version give one success; sales promotes a customer
only (H14).
"""
import threading
import json
import os
import pathlib
import subprocess
import sys
from urllib.parse import unquote, urlparse
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_bp.application import use_cases as uc # noqa: E402
from brick_bp.domain.partner import bp_id_for # noqa: E402
from brick_bp.infrastructure.tables import PartnerRepository, md # noqa: E402
def person(email, permissions):
return {"sub": email, "email": email, "roles": ["steward"], "permissions": permissions,
"auth_source": "pool"}
STEWARD = person("sam@co.example", ["bp:partner:write", "bp:family:write"])
CRM = person("crm@co.example", ["bp:partner:promote"])
STEWARD_PROMOTE = person("sam@co.example", ["bp:partner:write", "bp:partner:promote"])
HR = person("hr-service@co.example", ["bp:person:sync", "bp:person:read"])
ANYONE = person("ann@co.example", [])
CARD = {"person_id": "812", "legal_name": "Anna Berg", "preferred_name": "Anna", "work_email": "anna@co.example",
"work_phone": "+44 20 7946 0000", "roles": ["employee"]}
OLD_PARTNER = """CREATE TABLE business_partner (
bp_id BIGINT NOT NULL, bp_key VARCHAR NOT NULL, legal_name VARCHAR NOT NULL, country VARCHAR NOT NULL,
tax_id VARCHAR NOT NULL, domain VARCHAR NOT NULL, status VARCHAR NOT NULL, created_at TIMESTAMP,
updated_at TIMESTAMP, PRIMARY KEY (bp_id))"""
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 migrate_from_0_1(tables) -> bool:
"""A bucket as 0.1.1 left it — then the real erp-migrate (plan, apply, VERIFY) of this schema.sql."""
lc.drop_all(tables)
lc.pg(OLD_PARTNER)
lc.pg(f"INSERT INTO business_partner (bp_id, bp_key, legal_name, country, tax_id, domain, status) "
f"VALUES ({bp_id_for('old.example')}, 'old.example', 'Old Co', 'GB', '', '', 'active')")
# erp_platform.db reads the pg password from DODIL_TOKEN locally: from `dodil data connect` (the CLI login)
# into this process only, never printed. livecheck's own CLI calls leave it out (see livecheck._cli_env).
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).stdout)["pg"])
os.environ["DODIL_TOKEN"] = unquote(u.password or "")
from erp_platform import migrate
return migrate.migrate(ROOT / "src" / "brick_bp" / "schema.sql", {"bp": uc.BP.version}) == 0
def main() -> int:
tables = [t.name for t in md.sorted_tables]
lc.check("the additive migration from a 0.1 bucket applies and verifies (erp-migrate)", migrate_from_0_1(tables))
uc.repository = lc.repository(PartnerRepository)
old = uc.get_partner(ANYONE, bp_id_for("old.example"))
lc.check("an organisation written before 0.2.0 reads as an organisation, unchanged",
old["kind"] == "organisation" and old["legal_name"] == "Old Co" and "work_email" not in old, old)
lc.pg(f"DELETE FROM business_partner WHERE bp_id = {bp_id_for('old.example')}")
a = uc.upsert_partner(STEWARD, "Acme.com", "Acme AB", "se", tax_id="SE5560123401", roles=["customer"])
first = lc.rows(f"SELECT created_at FROM business_partner WHERE bp_id = {a['bp_id']}")[0]["created_at"]
b = uc.upsert_partner(STEWARD, "acme.com", "Acme AB", "SE", roles=["supplier"])
# bp_id read as text: the CLI's JSON passes BIGINT through a float (see erp_platform.livecheck)
stored = lc.rows("SELECT CAST(bp_id AS VARCHAR) AS bp_id, bp_key, tax_id, created_at FROM business_partner")
lc.check("BP-02 the same company imported twice is one row with the derived id, key normalised",
len(stored) == 1 and int(stored[0]["bp_id"]) == bp_id_for("acme.com") == a["bp_id"] == b["bp_id"]
and stored[0]["bp_key"] == "acme.com", stored)
lc.check("a re-import without a tax id keeps it; created_at kept (tables#417 workaround)",
stored[0]["tax_id"] == "SE5560123401" and stored[0]["created_at"] == first, stored)
lc.check("BP-01 one partner, two roles",
sorted(r["role"] for r in lc.rows(f"SELECT role FROM bp_role WHERE bp_id = {a['bp_id']}"))
== ["customer", "supplier"])
uc.set_partner_status(STEWARD, a["bp_id"], "blocked", "unpaid invoices")
again = uc.upsert_partner(STEWARD, "acme.com", "Acme AB", "SE")
lc.check("a re-import never unblocks", again["status"] == "blocked", again["status"])
ids = {k: uc.upsert_partner(STEWARD, k, n, c)["bp_id"] for k, n, c in (
("acme-group", "Acme Group", "GB"), ("acme-uk", "Acme UK", "GB"), ("acme-de", "Acme DE", "DE"),
("acme-berlin", "Acme DE Berlin", "DE"), ("frame-co", "Framework Co", "FR"))}
for child, parent, rel in (("acme-uk", "acme-group", "subsidiary_of"),
("acme-de", "acme-group", "subsidiary_of"),
("acme-berlin", "acme-de", "subsidiary_of"),
("frame-co", "acme-group", "framework_governs")):
uc.link_partners(STEWARD, ids[child], ids[parent], rel)
fam = uc.get_family(STEWARD, ids["acme-group"])
lc.check("BP-03 the recursive walk on DataK3 is typed: 3 subsidiaries, the framework stays out",
sorted(m["legal_name"] for m in fam["below"]) == ["Acme DE", "Acme DE Berlin", "Acme UK"], fam)
cyc = lc.refused(uc.link_partners, STEWARD, ids["acme-group"], ids["acme-berlin"])
lc.check("a cycle through the stored graph is refused", cyc and cyc[0] == 409 and "cycle" in cyc[1], cyc)
uc.unlink_partners(STEWARD, ids["acme-berlin"], ids["acme-de"])
lc.check("unlinking removes exactly that edge",
sorted(m["legal_name"] for m in uc.get_family(STEWARD, ids["acme-group"])["below"])
== ["Acme DE", "Acme UK"])
p = uc.promote_partner(CRM, "newco.example", legal_name="NewCo Ltd", country="GB")
no = lc.refused(uc.promote_partner, CRM, "newco.example", role="supplier")
lc.check("H14 with bp:partner:promote alone only the customer role is given",
no == (403, "missing permission: bp:partner:write"), no)
p2 = uc.promote_partner(STEWARD_PROMOTE, "newco.example", role="supplier")
lc.check("BP-04 promote creates, then reuses and adds the role",
p["bp_id"] == p2["bp_id"] and p2["roles"] == ["customer", "supplier"], p2)
# ── 0.3.0: the optimistic version on DataK3 ───────────────────────────────────────────
v = uc.get_partner(ANYONE, p["bp_id"])["version"]
uc.set_partner_status(STEWARD, p["bp_id"], "blocked", "check", version=v)
stale = refused(uc.set_partner_status, STEWARD, p["bp_id"], "active", "check", version=v)
lc.check("C3 a stale version is 409 'changed since you opened it' and writes nothing",
stale and stale[0] == 409 and "changed since you opened it" in stale[1]
and uc.get_partner(ANYONE, p["bp_id"])["status"] == "blocked", stale)
v = uc.get_partner(ANYONE, p["bp_id"])["version"]
out = []
def race(status):
try:
r = refused(uc.set_partner_status, STEWARD, p["bp_id"], status, "race", version=v)
out.append(("refused", str(r[0])) if r else status)
except Exception as e: # noqa: BLE001 — counted
out.append(("error", type(e).__name__, str(e)[-200:]))
ts = [threading.Thread(target=race, args=(s,)) for s in ("active", "archived", "active", "archived")]
for t in ts:
t.start()
for t in ts:
t.join()
wins = [o for o in out if not isinstance(o, tuple)]
lc.check("C3 four stewards racing from one version on DataK3: one success, the rest 409",
len(wins) == 1 and all(o == ("refused", "409") for o in out if isinstance(o, tuple))
and uc.get_partner(ANYONE, p["bp_id"])["version"] == v + 1, out)
hist = [e["action"] for e in uc.partner_history(STEWARD, a["bp_id"])["events"]]
lc.check("history records each change once", hist == ["created", "updated", "status blocked"], hist)
org_row = lc.rows(f"SELECT kind, work_email FROM business_partner WHERE bp_id = {a['bp_id']}")[0]
lc.check("an organisation's save never names the person columns (kind stays NULL)",
org_row["kind"] is None and org_row["work_email"] is None, org_row)
# ── person partners ────────────────────────────────────────────────────────────────────
psc = uc.upsert_partner(STEWARD, "anna-consulting.example", "Anna Consulting Ltd", "GB", roles=["supplier"])
p = uc.sync_person(HR, {**CARD, "roles": ["worker"], "works_for": psc["bp_id"]})
again = uc.sync_person(HR, {**CARD, "roles": ["worker"], "works_for": psc["bp_id"]})
stored = lc.rows(f"SELECT kind, bp_key, legal_name, work_email, tax_id FROM business_partner "
f"WHERE bp_id = {bp_id_for('person:812')}")
lc.check("BP-05 one person partner, keyed person:<person_id>, its bp_id derived; a second sync changes nothing",
p == again and p["bp_id"] == bp_id_for("person:812") and len(stored) == 1
and stored[0]["kind"] == "person" and stored[0]["bp_key"] == "person:812"
and [e["action"] for e in uc.partner_history(HR, p["bp_id"])["events"]] == ["person created"], stored)
fam = uc.get_family(HR, psc["bp_id"], "works_for")
lc.check("BP-05 works_for: the freelancer is below their company on the typed walk",
[m["bp_id"] for m in fam["below"]] == [p["bp_id"]], fam)
priv = lc.refused(uc.sync_person, HR, {**CARD, "personal_email": "anna@home.example"})
lc.check("BP-05 a private field is refused in the rule's words",
priv and priv[0] == 422 and "a name and a work contact only" in priv[1], priv)
hr_only = lc.refused(uc.upsert_partner, STEWARD, "person:812", "X", "GB")
lc.check("BP-05 bp:partner:write cannot write a person", hr_only and hr_only[0] == 409, hr_only)
listed = [x["bp_id"] for x in uc.find_partners(ANYONE, name="anna")["partners"]]
lc.check("BP-06 without bp:person:read the person is absent from the list and 404 by id",
listed == [psc["bp_id"]] and lc.refused(uc.get_partner, ANYONE, p["bp_id"])[0] == 404
and uc.get_family(ANYONE, psc["bp_id"], "works_for")["below"] == [], listed)
lc.check("BP-06 with bp:person:read both are listed",
{x["bp_id"] for x in uc.find_partners(HR, name="anna")["partners"]} == {psc["bp_id"], p["bp_id"]})
gone = uc.sync_person(HR, {"person_id": "812"}, anonymise=True)
gone2 = uc.sync_person(HR, {"person_id": "812"}, anonymise=True)
row = lc.rows(f"SELECT legal_name, preferred_name, work_email, work_phone, anonymised_at FROM business_partner "
f"WHERE bp_id = {p['bp_id']}")[0]
lc.check("BP-07 anonymise: the stored name and work contact are the placeholder; the id, roles and link stay; "
"twice changes nothing",
gone == gone2 and row["legal_name"] == "Erased person" and row["work_email"] == ""
and row["work_phone"] == "" and row["anonymised_at"] is not None and gone["roles"] == ["worker"]
and [m["bp_id"] for m in uc.get_family(HR, psc["bp_id"], "works_for")["below"]] == [p["bp_id"]], row)
hist = lc.rows(f"SELECT detail FROM bp_events WHERE bp_id = {p['bp_id']}")
lc.check("BP-07 the history held no name or contact to erase", not any("Anna" in (h["detail"] or "")
or "anna@" in (h["detail"] or "") for h in hist), hist)
return lc.finish(tables)
if __name__ == "__main__":
raise SystemExit(main())