dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
scripts/live_check.py
"""ONE live check: the brick alone on a scratch DataK3 bucket (org dodil) over the REAL pg wire, with real transactions
(erp_platform.livecheck.engine). The schema is applied from ALL the migrations IN ORDER (00001_init, 00002_slice_c), then the stories whose
correctness depends on DataK3's behaviour, not on Python logic:
- two translate runs, then two elimination runs, racing on ONE group period: each 200 or 409, never a 500, and exactly one run written
(one claim per group period, task and run; one set of entries)
- a repeat run writes nothing
- two manual entries approved at once: the entry numbers are distinct and above the high-water mark (the claim row; either racer may lose at
commit -> 409)
- the sign-off racing a re-collection: no 500, and a signed period never carries a digest other than the one signed
- the sign-off history is append-only (the signed row is never rewritten by the reopen) and read back with aware timestamps
- the MCP tool count
gl is the spec-true stand-in of tests/fakes.py (FakeGl): every call is checked against gl's signed spec (ERP_METHOD_DIR) as the fake makes it. The
token is the logged-in CLI session's, taken by `dodil data connect` inside erp_platform.livecheck and never printed. Every request's duration is
logged; the slowest three are reported. The exit code is erp_platform.livecheck.finish's.
ERP_METHOD_DIR=../erp-method PATH=$PATH:~/.local/bin BUCKET=w2conlab DODIL_ORG=dodil python scripts/live_check.py
"""
import os
import pathlib
import re
import sys
import threading
import time
from datetime import datetime, timezone
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / "src"), str(ROOT / "tests")]
os.environ.setdefault("BUCKET", "w2conlab")
os.environ.setdefault("DODIL_ORG", "dodil")
os.environ.setdefault("ERP_METHOD_DIR", str(ROOT.parents[0] / "erp-method"))
import conftest as C # noqa: E402 (personas and the two doors; its fixtures are not used)
_timings: list = []
_door = C.api
def timed(method, path, headers, body=None):
t0 = time.perf_counter()
out = _door(method, path, headers, body)
_timings.append(((time.perf_counter() - t0) * 1000, method, re.sub(r"/groups/GRP/group-periods/[\d-]+", "/groups/GRP/group-periods/P", path), out[0]))
return out
C.api = timed # before the helpers below bind `api` by name
import contract_checks # noqa: E402
import test_slice_b1 as B1 # noqa: E402
import test_slice_b2 as B2 # noqa: E402
from erp_platform import livecheck as lc # noqa: E402
from erp_platform import work # noqa: E402
from fakes import FakeGl, FakeStore # noqa: E402
from world import ACC, CTRL, OWNER, P, ok # noqa: E402
from brick_consolidation.application import use_cases as uc # noqa: E402
from brick_consolidation.infrastructure import seams # noqa: E402
from brick_consolidation.infrastructure.tables import md # noqa: E402
BASE = f"/groups/GRP/group-periods/{P}"
TABLES = ["cons_schema_migrations"] + list(md.tables) # the migrations ledger is created by 00001_init, not by the model
def apply_migrations() -> int:
n = 0
for d in sorted((ROOT / "src" / "brick_consolidation" / "migrations").iterdir()):
for stmt in re.split(r";\s*\n", (d / "up.sql").read_text()):
body = "\n".join(x for x in stmt.splitlines() if not x.strip().startswith("--")).strip().rstrip(";")
if body:
lc.pg(body)
n += 1
return n
def n_rows(table, where="TRUE"):
return int(lc.rows(f"SELECT COUNT(*) AS n FROM {table} WHERE {where}")[0]["n"])
def race(*calls):
"""Run the calls at once; each is (method, path, headers, body). -> their statuses (a crash is a result too)."""
out = [None] * len(calls)
def go(i, c):
try:
out[i] = C.api(*c)[0]
except Exception as exc: # noqa: BLE001 - a crash is a failure of the story, reported by name
out[i] = type(exc).__name__
ts = [threading.Thread(target=go, args=(i, c)) for i, c in enumerate(calls)]
[t.start() for t in ts]
[t.join() for t in ts]
return out
def step(name, who=ACC, body=None):
return ("POST", f"{BASE}/{name}", who, body or {})
def period():
return lc.rows(f"SELECT status, run_no, tb_digest FROM cons_group_period WHERE group_id = 'GRP' AND period_id = '{P}'")[0]
def main():
eng = lc.engine()
lc.drop_all(TABLES)
print(f"migrations applied: {apply_migrations()} statements on {os.environ['BUCKET']}")
uc.repository = lambda: uc._on_engine(eng) # the brick's own commit-time conflict handling, on a real transaction (DataK3 decides at COMMIT)
work.set_clock(at=datetime(2026, 11, 20, 9, 0, tzinfo=timezone.utc), freeze=True)
gl = FakeGl()
seams.ports.gl, seams.ports.files = gl, FakeStore()
with uc.repository() as r:
r.seed()
# ── the group closed up to elimination, on the live bucket ────────────────────────────────────
B2.ready_world(gl)
lc.check("the group is eliminated on the live bucket", period()["status"] == "eliminated" and int(period()["run_no"]) == 1, period())
# ── a repeat run writes nothing ───────────────────────────────────────────────────────────────
tables = ("cons_translated_data", "cons_journal", "cons_journal_line", "cons_run_claim", "cons_collection", "cons_reported_data", "cons_number_series")
before = {t: n_rows(t) for t in tables}
again = ok(C.api(*step("run-eliminations")))
lc.check("a repeat elimination run writes nothing", again["result"] == {"changed": False} and {t: n_rows(t) for t in tables} == before, again["result"])
# ── two runs racing on ONE group period (run 2, after gl's figures changed) ───────────────────
def change_gl(extra):
rows = [(r["group_account_id"], r["ytd_amount"], r["period_amount"], r["partner_company"]) for r in gl.figures[("DE01", P)]["rows"]
if r["group_account_id"] not in ("G-6500", "G-2000") or r["partner_company"]]
gl.set_rows("DE01", P, rows + extra)
change_gl([("G-6500", "5.00", "5.00"), ("G-2000", "-5.00", "-5.00")])
ok(C.api(*step("collect-group-data")))
lc.check("a changed gl digest takes run 2 and the period is collected again", int(period()["run_no"]) == 2 and period()["status"] == "collected", period())
got = race(step("translate-currencies"), step("translate-currencies"))
lc.check("two translate runs racing: 200 or 409, never a 500, and one succeeded", all(s in (200, 409) for s in got) and 200 in got, got)
if 409 in got:
ok(C.api(*step("translate-currencies"))) # the loser retries: the run is already written, nothing is written twice
lc.check("one claim for (translate, run 2) and the period is translated", n_rows("cons_run_claim", "task = 'translate' AND run_no = 2") == 1
and period()["status"] == "translated", (n_rows("cons_run_claim", "task = 'translate' AND run_no = 2"), period()))
got = race(step("run-eliminations"), step("run-eliminations"))
lc.check("two elimination runs racing: 200 or 409, never a 500, and one succeeded", all(s in (200, 409) for s in got) and 200 in got, got)
if 409 in got:
ok(C.api(*step("run-eliminations")))
sets = lc.rows("SELECT document_type, COUNT(*) AS n FROM cons_journal WHERE run_no = 2 AND status = 'posted' AND document_type <> 'manual' AND reverses_id = '' "
"GROUP BY document_type")
lc.check("one set of elimination entries for run 2 (each document type once) and run 1's reversed, never deleted",
sets and all(int(s["n"]) == 1 for s in sets) and n_rows("cons_journal", "run_no = 1 AND status = 'reversed'") >= 1
and n_rows("cons_run_claim", "task = 'eliminate' AND run_no = 2") == 1, (sets, n_rows("cons_run_claim", "task = 'eliminate' AND run_no = 2")))
# ── entry numbers racing ──────────────────────────────────────────────────────────────────────
B1.add_rules(("G-6600", "average"))
ids = [ok(B1.propose(key))["cons_journal_id"] for key in ("E1", "E2")]
top = int(lc.rows("SELECT COALESCE(MAX(entry_no), 0) AS n FROM cons_number_series")[0]["n"])
approve = lambda k: ("POST", f"/cons-journals/{k}/approve-cons-entry", CTRL, {}) # noqa: E731
got = race(approve(ids[0]), approve(ids[1]))
lc.check("two approvals racing for the next entry number: 200 or 409, never a 500", all(s in (200, 409) for s in got) and 200 in got, got)
for k, s in zip(ids, got):
if s == 409:
ok(C.api(*approve(k))) # the loser retries and is numbered
nos = [int(x["entry_no"]) for x in lc.rows("SELECT entry_no FROM cons_journal WHERE document_type = 'manual' AND status = 'posted'")]
claims = [int(x["entry_no"]) for x in lc.rows("SELECT entry_no FROM cons_number_series")]
lc.check("the entry numbers are distinct, above the high-water mark, and match the claim rows",
len(nos) == len(set(nos)) == 2 and min(nos) > top and sorted(claims) == sorted(set(claims)) and set(nos) <= set(claims), (nos, top, claims))
# ── the sign-off racing a re-collection ───────────────────────────────────────────────────────
ok(C.api(*step("translate-currencies")) if period()["status"] == "collected" else (200, {}))
if period()["status"] == "translated":
ok(C.api(*step("run-eliminations")))
change_gl([("G-6500", "6.00", "6.00"), ("G-2000", "-6.00", "-6.00")])
got = race(step("sign-off-group-close", CTRL), step("collect-group-data"))
final = period()
signed = lc.rows(f"SELECT tb_digest FROM cons_signoff WHERE group_id = 'GRP' AND period_id = '{P}' AND action = 'signed'")
lc.check("sign-off racing a re-collection: no 500, and a signed period never carries a digest other than the one signed",
all(s in (200, 409, 422) for s in got) and (final["status"] != "signed" or (signed and signed[-1]["tb_digest"] == final["tb_digest"])), (got, final, signed))
while period()["status"] != "signed": # whichever way the race went, bring the period to signed (recollect, translate, eliminate, sign)
status = period()["status"]
name = {"open": "collect-group-data", "collected": "translate-currencies", "translated": "run-eliminations", "eliminated": "sign-off-group-close"}[status]
ok(C.api(*step(name, CTRL if name.startswith("sign") else ACC)))
# ── the sign-off history is append-only and read back with aware timestamps ──────────────────
history = ok(C.api("GET", f"{BASE}/signoffs", CTRL))["signoffs"]
first = dict(history[-1])
lc.check("the signed history row reads back with an aware timestamp", first["action"] == "signed" and re.search(r"(\+00:00|Z)$", str(first["at"])), first)
ok(C.api("POST", f"{BASE}/reopen-group-period", B2.CTRL2, {"reason": "late posting in DE01"}))
after = ok(C.api("GET", f"{BASE}/signoffs", CTRL))["signoffs"]
lc.check("the reopen appends a row; the signed row is never rewritten", len(after) == len(history) + 1 and after[len(history) - 1] == first
and after[-1]["action"] == "reopened" and re.search(r"(\+00:00|Z)$", str(after[-1]["at"])), after[-2:])
raw = lc.rows(f"SELECT seq, action FROM cons_signoff WHERE group_id = 'GRP' AND period_id = '{P}' ORDER BY seq")
lc.check("the rows on the wire are numbered without gaps", [int(r["seq"]) for r in raw] == list(range(1, len(raw) + 1)), raw)
# ── the agent door ───────────────────────────────────────────────────────────────────────────
everyone = C.person("all@live", sorted({u.permission for u in uc.CONS.use_cases if u.permission}))
names = C.tools(everyone)
expected = sorted(u.name for u in uc.CONS.use_cases if u.tool)
lc.check(f"MCP: {len(names)} tools, one per tool-enabled use case", names == expected and names, (len(names), len(expected)))
print(f"MCP tools: {len(names)}")
lc.check("every seam call kept to the callee's signed spec", not contract_checks.CONTRACT.violations, contract_checks.CONTRACT.violations[:3])
slow = sorted(_timings, reverse=True)[:3]
print(f"requests: {len(_timings)}; slowest three:")
for ms, method, path, status in slow:
print(f" {ms:8.0f} ms {method} {path} -> {status}")
_ = OWNER
if __name__ == "__main__":
try:
main()
finally: # replaces an exception in flight: finish() records a crash as a failure
raise SystemExit(lc.finish(TABLES))