dodil/erp-brick-sales-order
DODIL ERP brick: sales order
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 .. 00006_slice_c), then the stories whose
correctness depends on DataK3's behaviour, not on Python logic:
- an order confirmed; two confirms racing for the next order number (the claim row: on DataK3 EITHER racer may lose at commit -> 409,
never a 500, never two numbers)
- SO-36: the re-promise run twice the same business day (the second writes nothing); a re-promise racing a cancel of the same order
(one wins, the other is refused, nothing half-written), and once more with the cancel forced into the re-promise's window
- SO-41/42: two credit requests racing for one invoice line (so_credit_claim): the total never exceeds the invoiced net
- SO-47: one deposit deducted by two racing invoice requests (so_deposit_deduction): deducted once
- SO-39: a block set then released, the appended rows read back
- SO-33: the rebate accrual run twice, and two runs racing: one journal per document
- the MCP tool count
Neighbours (product, tax, payments, invoicing, partners, inventory, ledger, purchasing) are the spec-true stand-ins of tests/fakes.py,
registered in process; each call is checked against the callee's signed spec (ERP_METHOD_DIR) as the fakes make it. The token is the logged-in CLI
session's, taken by `dodil data connect` inside erp_platform.livecheck and never printed. The exit code is erp_platform.livecheck.finish's.
ERP_METHOD_DIR=../erp-method PATH=$PATH:~/.local/bin BUCKET=w2solab DODIL_ORG=dodil python scripts/live_check.py
"""
import os
import pathlib
import re
import sys
import threading
from datetime import date
from decimal import Decimal as D
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / "src"), str(ROOT / "tests")]
os.environ.setdefault("BUCKET", "w2solab")
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)
import contract_checks # noqa: E402
import test_slice_b as B # noqa: E402 (scenario helpers: they run against whatever `use_cases.repository` is)
import test_slice_c as SC # noqa: E402
from erp_platform import livecheck as lc # noqa: E402
from erp_platform import work # noqa: E402
from fakes import ( # noqa: E402
FakeGl,
FakeInventory,
FakeInvoicing,
FakeMaps,
FakeOdoo,
FakePartners,
FakePayments,
FakeProduct,
FakePurchasing,
FakeTax,
)
from brick_sales_order.application import promising # noqa: E402
from brick_sales_order.application import use_cases as uc # noqa: E402
from brick_sales_order.infrastructure import seams # noqa: E402
from brick_sales_order.infrastructure.tables import md # noqa: E402
def apply_migrations() -> int:
n = 0
for d in sorted((ROOT / "src" / "brick_sales_order" / "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
class N: # the neighbours, as the tests wire them
pass
def main():
eng = lc.engine()
lc.drop_all(list(md.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
uc.SO.idempotency_connection = eng.begin
work.set_clock(at=C.datetime(2026, 11, 20, 9, 0, tzinfo=C.timezone.utc), freeze=True)
N.product, N.tax, N.payments, N.invoicing, N.partners = FakeProduct(), FakeTax(), FakePayments(), FakeInvoicing(), FakePartners()
N.maps, N.odoo, N.gl, N.purchasing = FakeMaps(), FakeOdoo(), FakeGl(), FakePurchasing()
inv = FakeInventory()
for name in ("product", "tax", "payments", "invoicing", "partners", "maps", "odoo", "gl", "purchasing"):
setattr(seams.ports, name, getattr(N, name))
seams.ports.inventory = inv
with uc.repository() as r:
r.seed()
REP, ADMIN, BILLER, SYSTEM, FULFILLER = C.REP, C.ADMIN, C.BILLER, C.SYSTEM, C.FULFILLER
must = C.must
# ── an order, and two confirms racing for the next order number ───────────────────────────────
one = C.confirm(C.hand(REP))
lc.check("an order is drafted, priced and confirmed with a number", one["lifecycle"] == "confirmed" and one["order_no"], one.get("order_no"))
a, b = C.hand(REP), C.hand(REP)
path = lambda o: ("POST", f"/sales-orders/{o['sales_order_id']}/confirm", REP, {}) # noqa: E731
got = race(path(a), path(b))
lc.check("two confirms racing for one order number: 200 or 409, never a 500", all(s in (200, 409) for s in got) and 200 in got, got)
for o, s in zip((a, b), got):
if s == 409:
C.confirm(o) # the loser retries and is numbered
nos = [x["order_no"] for x in lc.rows("SELECT order_no FROM so_order WHERE order_no IS NOT NULL AND order_no <> ''")]
claims = n_rows("so_order_no_claim")
lc.check("every order number is held once: the numbers are distinct and match the claim rows", len(nos) == len(set(nos)) == claims == 3, (nos, claims))
# ── SO-39: a block set then released, read back ───────────────────────────────────────────────
for code, kind in (("CHECK_TERMS", "delivery"), ("PRICE_DISPUTE", "billing")):
must(C.api("POST", "/block-reasons", ADMIN, {"block_reason_code": code, "kind": kind, "name": code}))
o = C.confirm(C.hand(REP, lines=B.WID))
oid = o["sales_order_id"]
must(C.api("POST", f"/sales-orders/{oid}/blocks", ADMIN, {"kind": "delivery", "reason_code": "CHECK_TERMS", "note": "terms to check"}))
held = C.refused(C.deliver(o, 1, 10, "D1"))
lc.check("a delivery block stops delivery", "delivery_block" in held, held)
must(C.api("POST", f"/sales-orders/{oid}/blocks/release", ADMIN, {"kind": "delivery", "note": "terms checked"}))
lc.check("after the release the delivery is recorded", C.deliver(o, 1, 10, "D1", C.INVENTORY, source="inventory")[0] == 200)
blocks = C.get(oid)["blocks"]
lc.check("the appended block rows read back in order (set, release)", [(x["block_no"], x["event"]) for x in blocks] == [(1, "set"), (2, "release")], blocks)
stamps = lc.rows(f"SELECT \"at\" FROM so_order_block WHERE sales_order_id = '{oid}'")
print(f" block rows on the wire: {len(stamps)}, sample at = {stamps[0]['at']}")
# ── SO-33: the rebate accrual, twice and racing ───────────────────────────────────────────────
B.active_agreement()
N.invoicing.add_billed("INV-2027-0101", 7311, "2027-02-03", [{"net_amount": "10000.00"}])
first = B.accrue("2027-02-04", "2027-02-03")
again = B.accrue("2027-02-04", "2027-02-03")
lc.check("the nightly accrual writes the line and the journal once", first["agreements"][0]["accrued"] == 1 and first["posted"] == 1
and again["agreements"][0]["accrued"] == 0 and again["posted"] == 0 and len(N.gl.journals) == 1, (first, again))
N.invoicing.add_billed("INV-2027-0102", 7311, "2027-02-05", [{"net_amount": "2000.00"}])
with C.on_day("2027-02-06T09:00:00+00:00"):
got = race(*[("POST", "/rebate-agreements/accrue", SYSTEM, {"for_date": "2027-02-05"})] * 2)
lc.check("two accrual runs racing: 200 or 409, never a 500", all(s in (200, 409) for s in got) and 200 in got, got)
if 409 in got:
B.accrue("2027-02-06", "2027-02-05") # the loser's retry completes whatever is left, writing nothing twice
docs = lc.rows("SELECT document_id, journal_id, posting_status FROM so_rebate_accrual ORDER BY document_id")
keys = sorted(j["source_document"] for j in N.gl.journals.values())
lc.check("one accrual row and one journal per document", [d["document_id"] for d in docs] == ["INV-2027-0101", "INV-2027-0102"]
and len(keys) == len(set(keys)) == 2 and all(d["posting_status"] == "posted" for d in docs), (docs, keys))
# ── SO-41/42: two credit requests racing for one invoice line ─────────────────────────────────
w = SC.original(N, inv)
c1, c2 = C.confirm(SC.memo(N, price="0.80")), C.confirm(SC.memo(N, price="0.80")) # each 19.20 of the invoice line's 28.80
ctl = SC.CONTROLLER
approve = lambda x: ("POST", f"/sales-orders/{x['sales_order_id']}/approve-memo", ctl, {}) # noqa: E731
got = race(approve(c1), approve(c2))
held = D(str(lc.rows(f"SELECT COALESCE(SUM(amount), 0) AS t FROM so_credit_claim WHERE claim_key = 'inv:{SC.INV40}:1' AND kind = 'credit'")[0]["t"]))
print(" credit race:", got, "| a repeat says:", C.api(*approve(c1))[1])
lc.check("two credits racing for one invoice line: one 200, the other refused, never a 500", all(s in (200, 409) for s in got) and got.count(200) == 1, got)
lc.check("the credits held on the line never exceed the invoiced net (28.80), and invoicing got one credit note",
held == D("19.20") and len(N.invoicing.credits) == 1, (held, len(N.invoicing.credits)))
del w
# ── SO-47: one deposit, two racing invoice requests ───────────────────────────────────────────
d = C.confirm(C.hand(REP, lines=SC.WIDGETS1000))
d1 = SC.deposit(N, d, "5000.00")
must(C.deliver(d, 1, 400, "DN-1", C.INVENTORY, source="inventory")) # inventory is installed: a stocked line is delivered only by it
before = set(N.invoicing.drafts) # the deposit's own draft is not the race's
got = race(*[("POST", f"/sales-orders/{d['sales_order_id']}/request-invoice", BILLER, {})] * 2)
taken = D(str(lc.rows(f"SELECT COALESCE(SUM(amount), 0) AS t FROM so_deposit_deduction WHERE deposit_invoice_id = '{d1}'")[0]["t"]))
drafts = [k for k in N.invoicing.drafts if k.startswith(f"so:{d['sales_order_id']}:") and k not in before]
lc.check("two invoice requests racing: no 500, and the deposit is deducted once (5,000 / 10,000 x 4,000 = 2,000.00, not 4,000.00)",
all(s in (200, 409, 422) for s in got) and 200 in got and taken == D("2000.00") and len(drafts) == 1, (got, taken, drafts))
# (SO-36 runs last: its partner rows make the payer 7400, and the credit and deposit stories bill 7311)
# ── SO-36: re-promise twice, then racing a cancel ─────────────────────────────────────────────
o1 = B.promised_240(inv, N)
oid1 = o1["sales_order_id"]
B.purchase_moved(inv)
with C.on_day(B.DAY2):
r1 = B.reschedule()
sched, tasks, version = n_rows("so_schedule_line"), n_rows("so_task"), C.get(oid1)["version"]
inv.supply["JUICE-075"] = [(date(2026, 11, 5), D("300"), None), (date(2026, 11, 17), D("720"), None)]
r2 = B.reschedule()
lc.check("the re-promise run twice the same business day: the second writes nothing", oid1 in r1["re_promised"] and r2["re_promised"] == []
and r2["tasks"] == 0 and C.get(oid1)["version"] == version
and n_rows("so_schedule_line") == sched and n_rows("so_task") == tasks, (r1, r2, sched, tasks))
def mk():
with C.on_day(B.DAY1):
return C.confirm(C.hand(REP, lines=[{"product_id": "JUICE-075", "quantity": "240", "unit_code": "C62"}], requested_delivery_on="2026-11-04"))
o2 = mk()
oid2 = o2["sales_order_id"]
inv.supply["JUICE-075"] = [(date(2026, 11, 5), D("90"), None), (date(2026, 11, 18), D("720"), None)]
with C.on_day("2026-11-06T09:00:00+00:00"):
got = race(("POST", "/sales-orders/reschedule", SYSTEM, {}), ("POST", f"/sales-orders/{oid2}/cancel", C.ADMIN, {"reason": "gone"}))
final = C.get(oid2)
versions = {int(x["schedule_version"]) for x in lc.rows(f"SELECT schedule_version FROM so_schedule_line WHERE sales_order_id = '{oid2}'")}
lc.check("a re-promise racing a cancel: each 200 or 409, the order cancelled, its schedule versions whole (no half-written version)",
all(s in (200, 409) for s in got) and final["lifecycle"] == "cancelled" and versions == set(range(1, max(versions) + 1))
and final["schedule_version"] == max(versions), (got, final["lifecycle"], versions))
o3 = mk()
oid3 = o3["sales_order_id"]
real, fired = promising.propose, []
def propose_then_cancel(repo, o, *a, **kw):
out = real(repo, o, *a, **kw)
if o.sales_order_id == oid3 and not fired:
fired.append(1)
must(C.api("POST", f"/sales-orders/{oid3}/cancel", C.ADMIN, {"reason": "gone"})) # another unit of work saves the order first
return out
promising.propose = propose_then_cancel
try:
inv.supply["JUICE-075"] = [(date(2026, 11, 8), D("90"), None), (date(2026, 11, 21), D("720"), None)]
with C.on_day("2026-11-09T09:00:00+00:00"):
res = B.reschedule()
finally:
promising.propose = real
after = C.get(oid3)
lc.check("a cancel landing inside the re-promise window: the re-promise loses on the order version and writes nothing", fired and oid3 not in res["re_promised"]
and after["lifecycle"] == "cancelled" and after["schedule_version"] == 1
and n_rows("so_task", f"sales_order_id = '{oid3}' AND kind = 'repromise'") == 0, (fired, res, after["lifecycle"], after["schedule_version"]))
# ── the agent door ───────────────────────────────────────────────────────────────────────────
everyone = C.person("all@live", sorted({u.permission for u in uc.SO.use_cases if u.permission}))
names = C.tools(everyone)
expected = sorted(u.name for u in uc.SO.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])
if __name__ == "__main__":
try:
main()
finally: # replaces an exception in flight: finish() records a crash as a failure
raise SystemExit(lc.finish(list(md.tables)))