"""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 .. 00006), with an untyped product seeded
BEFORE 00006, then the stories whose correctness depends on DataK3's behaviour, not on Python logic:
- migration 00006 on a legacy untyped product: it ends typed (LEGACY / SERVICE), its behaviour unchanged
- two first GTINs racing on one unit: exactly one main, the other an additional one or a 409, never a 500
- a classify racing the activation: one wins, never an ACTIVE product in a class with a required value missing
- code-series claims racing: distinct codes, above the high-water mark, one claim row each
- dated rows read by business date (a price change), and aware timestamps (a unit marked used stays fixed)
- GET /product-sites (paged) and the Odoo migration in chunks of 100: run twice the second writes nothing; a run stopped between
chunks resumes where the committed chunks end. Their durations are logged.
- mark-unit-used's refusal body {gate: einvoice_unit_code, reason, fix}
- the MCP tool count
Inventory (GET /warehouses/{id}) is a spec-true stand-in registered in process: every call is checked against inventory's signed spec
(ERP_METHOD_DIR). 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=w2prdlab 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", "w2prdlab")
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"/products/[^/]+", "/products/P", path), out[0]))
return out
C.api = timed # before the helpers below bind `api` by name
import contract_checks # noqa: E402,F401 (fails loudly without an erp-method checkout)
from erp_platform import livecheck as lc # noqa: E402
from erp_platform import seam, work # noqa: E402
from erp_platform.contract import Recorder, _has_spec # noqa: E402
from fastapi import FastAPI, Response # noqa: E402
from test_audit_v4 import IMPORTER # noqa: E402
from brick_product.application import use_cases as uc # noqa: E402
from brick_product.infrastructure.tables import Repository, md # noqa: E402
CATALOGUE, PRICING, SALES, SELLING = C.CATALOGUE, C.PRICING, C.SALES, C.SELLING
PLANNER = C.as_role("planner")
TABLES = ["prod_schema_migrations"] + list(md.tables) # the migrations ledger is created by 00001_init, not by the model
MIGRATIONS = sorted((ROOT / "src" / "brick_product" / "migrations").iterdir())
def apply_migrations(dirs) -> int:
n = 0
for d in dirs:
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 gs1(body13: str) -> str:
"""A GTIN-14 with its right GS1 check digit."""
s = sum(int(d) * (3 if i % 2 == 0 else 1) for i, d in enumerate(body13))
return body13 + str((10 - s % 10) % 10)
def inventory_standin(rec):
app = FastAPI()
@app.get("/warehouses/{warehouse_id}")
def warehouse(warehouse_id: str, response: Response):
if warehouse_id != "WH1":
response.status_code = 404
return {"detail": "no such warehouse"}
return {"warehouse_id": warehouse_id, "company_code": "GB1", "status": "active"}
seam.register_local("inventory", app)
original = seam.Seam._call
method_dir = os.environ["ERP_METHOD_DIR"]
def call(self, method, path, actor, caller=None, **kw):
if self.installed and _has_spec(self.name, method_dir):
rec.verify(self.name, method, path, kw.get("json") if method in ("POST", "PUT", "PATCH") else kw.get("params"), "product", method_dir)
return original(self, method, path, actor, caller, **kw)
seam.Seam._call = call
def odoo_run(db, n, first=0):
return {"db": db, "cutover": "2026-11-01", "pack_codes": {"Pack of 24": "PK24"},
"uoms": [{"odoo_id": 1, "name": "Units"}, {"odoo_id": 2, "name": "Pack of 24", "relative_uom_id": 1, "relative_factor": "24.0", "rounding": "1"}],
"products": [{"odoo_id": 100 + i, "default_code": f"{db[-1].upper()}-{i:04d}", "type": "product", "name": f"Odoo {i}", "uom_id": 1, "weight": "0.5",
"uoms": [{"uom_id": 1}, {"odoo_id": 1000 + i, "uom_id": 2}]} for i in range(first, first + n)]}
def main():
eng = lc.engine()
lc.drop_all(TABLES)
must, api = C.must, C.api
work.set_clock(at=datetime(2026, 11, 20, 9, 0, tzinfo=timezone.utc), freeze=True)
# ── migration 00006 on a legacy untyped product ───────────────────────────────────────────────
print(f"migrations 00001-00005 applied: {apply_migrations(MIGRATIONS[:5])} statements on {os.environ['BUCKET']}")
for pid, kind, status in (("LEG-ACT", "goods", "active"), ("LEG-DRAFT", "goods", "draft"), ("LEG-SVC", "service", "draft")):
lc.pg("INSERT INTO prod_product (product_id, name, kind, base_unit_code, sales_unit_code, tax_category, billing_kind, billing_plan, invoicing_policy, status, version, "
f"created_at, updated_at) VALUES ('{pid}', 'Legacy {pid}', '{kind}', 'C62', 'C62', 'standard', 'one_off', 'single', 'ordered', '{status}', 1, "
"CAST('2026-09-01 09:00:00' AS TIMESTAMP), CAST('2026-09-01 09:00:00' AS TIMESTAMP))")
lc.pg(f"INSERT INTO prod_product_versions (product_id, version, saved_at) VALUES ('{pid}', 1, CAST('2026-09-01 09:00:00' AS TIMESTAMP))")
lc.pg("INSERT INTO prod_product (product_id, name, kind, base_unit_code, tax_category, billing_kind, billing_plan, invoicing_policy, status, version, "
"product_type_code, created_at, updated_at) VALUES ('TYP-1', 'Typed before 00006', 'goods', 'C62', 'standard', 'one_off', 'single', 'ordered', 'draft', 1, "
"'TRADING', CAST('2026-09-01 09:00:00' AS TIMESTAMP), CAST('2026-09-01 09:00:00' AS TIMESTAMP))")
print(f"migration 00006 applied: {apply_migrations(MIGRATIONS[5:])} statements")
typed = {r["product_id"]: (r["product_type_code"], r["sellable"], r["required_roles"]) for r in
lc.rows("SELECT product_id, product_type_code, sellable, required_roles FROM prod_product ORDER BY product_id")}
lc.check("00006 types the untyped rows: goods -> LEGACY, a service -> SERVICE, sellable, nothing required",
{k: v[0] for k, v in typed.items() if k != "TYP-1"} == {"LEG-ACT": "LEGACY", "LEG-DRAFT": "LEGACY", "LEG-SVC": "SERVICE"}
and all(str(v[1]).lower() in ("true", "1") and v[2] == "[]" for k, v in typed.items() if k != "TYP-1"), typed)
lc.check("00006 gives a typed row the copies from its type (the UPDATE that DataK3 cannot do as UPDATE ... FROM)",
typed["TYP-1"][0] == "TRADING" and str(typed["TYP-1"][1]).lower() in ("true", "1") and typed["TYP-1"][2] == '["sales", "purchase"]', typed["TYP-1"])
lc.check("00006 added the LEGACY type once", n_rows("prod_product_type", "product_type_code = 'LEGACY'") == 1)
uc.repository = lambda: uc._on_engine(eng) # the brick's own commit-time conflict handling, on a real transaction (DataK3 decides at COMMIT)
rec = Recorder()
inventory_standin(rec)
with uc.repository() as r:
r.seed()
got = must(api("GET", "/products/LEG-ACT", CATALOGUE))
lc.check("a legacy product reads as typed LEGACY, still active", got["product_type_code"] == "LEGACY" and got["status"] == "active", {k: got.get(k) for k in ("product_type_code", "status")})
lc.check("legacy drafts activate as before (nothing required)", all(must(api("POST", f"/products/{p}/activate-product", CATALOGUE, {}))["status"] == "active"
for p in ("LEG-DRAFT", "LEG-SVC")))
lc.check("a legacy active product renames as before", must(api("POST", "/products/LEG-ACT/rename-product", CATALOGUE, {"name": "Legacy renamed"}))["name"] == "Legacy renamed")
# ── two first GTINs racing on one unit (MUST-1) ───────────────────────────────────────────────
C.mk_product("BOX", name="Box", base_unit_code="C62", billing_kind="one_off", activate=False)
must(api("POST", "/products/BOX/set-units", CATALOGUE, {"sales_unit_code": "C62", "units_of": [
{"unit_code": "PK6", "contains_unit": "C62", "contains_quantity": "6", "einvoice_unit_code": "XPK"}]}))
must(api("POST", "/products/BOX/activate-product", CATALOGUE, {}))
ga, gb = gs1("0501234500001"), gs1("0400638133393")
ask = lambda g: ("POST", "/products/BOX/add-gtin", CATALOGUE, {"unit_code": "PK6", "gtin": g}) # noqa: E731
got = race(ask(ga), ask(gb))
lc.check("two first GTINs racing: 200 or 409, never a 500, one succeeded", all(s in (200, 409) for s in got) and 200 in got, got)
for g, s in zip((ga, gb), got):
if s == 409:
api(*ask(g)) # the loser asks again: now an additional barcode
kinds = {x["gtin"]: x["kind"] for x in lc.rows("SELECT gtin, kind FROM prod_gtin WHERE owner = 'BOX' AND unit_code = 'PK6'")}
mains = lc.rows("SELECT gtin FROM prod_gtin_main WHERE owner = 'BOX' AND unit_code = 'PK6'")
lc.check("exactly one main GTIN on the unit, the other additional, and the claim row names the main",
sorted(kinds.values()) == ["additional", "main"] and len(mains) == 1 and kinds[mains[0]["gtin"]] == "main", (kinds, mains))
lc.check("the claim rows are as the API reads them", {g["gtin"]: g["kind"] for u in must(api("GET", "/products/BOX", CATALOGUE))["units"]
if u["unit_code"] == "PK6" for g in u["gtins"]} == kinds)
# ── classify racing the activation ────────────────────────────────────────────────────────────
for code, body in (("MATERIAL", {"data_type": "list", "allowed_values": ["A2", "A4"]}), ("THREAD", {"data_type": "list", "allowed_values": ["M3", "M4"]})):
must(api("POST", f"/characteristics/{code}", CATALOGUE, body))
must(api("POST", "/product-classes/SCREWS", CATALOGUE, {"characteristics": [{"characteristic_code": "MATERIAL", "required": True},
{"characteristic_code": "THREAD", "required": True}]}))
outcomes, bad = [], []
for i in range(4):
pid = f"SCR-{i}"
C.mk_product(pid, name=pid, base_unit_code="C62", billing_kind="one_off", product_type_code="SERVICE", activate=False)
got = race(("POST", f"/products/{pid}/classify-product", CATALOGUE, {"class_code": "SCREWS", "values": [{"characteristic_code": "MATERIAL", "value": "A2"}]}),
("POST", f"/products/{pid}/activate-product", CATALOGUE, {}))
outcomes.append(got)
status = lc.rows(f"SELECT status FROM prod_product WHERE product_id = '{pid}'")[0]["status"]
member = n_rows("prod_classification", f"owner = '{pid}' AND class_code = 'SCREWS'")
thread = n_rows("prod_classification", f"owner = '{pid}' AND characteristic_code = 'THREAD'")
if status == "active" and member and not thread:
bad.append(pid)
lc.check("classify racing activate (4 rounds): 200 or 409, never a 500, and no active product in SCREWS without its THREAD",
all(s in (200, 409) for o in outcomes for s in o) and not bad, (outcomes, bad))
# ── code-series claims racing ─────────────────────────────────────────────────────────────────
must(api("POST", "/product-types/SPARE", CATALOGUE, {"name": "Spare parts", "code_format": "SP-{seq:4}", "required_roles": []}))
must(api("POST", "/products", CATALOGUE, {"product_type_code": "SPARE", "product_id": "SP-0007", "name": "Migrated spare", "base_unit_code": "C62", "tax_category": "standard",
"billing_kind": "one_off", "invoicing_policy": "ordered"}))
new = lambda n: ("POST", "/products", CATALOGUE, {"product_type_code": "SPARE", "name": f"Spare {n}", "base_unit_code": "C62", "tax_category": "standard", # noqa: E731
"billing_kind": "one_off", "invoicing_policy": "ordered"})
got = race(*[new(n) for n in range(4)])
lc.check("four adds racing for the next series number: 200 or 409, never a 500, one succeeded", all(s in (200, 409) for s in got) and 200 in got, got)
for n, s in enumerate(got):
if s == 409:
must(api(*new(n))) # the loser asks again and takes the next number
codes = [x["product_id"] for x in lc.rows("SELECT product_id FROM prod_product WHERE product_type_code = 'SPARE' ORDER BY product_id")]
claims = [int(x["number"]) for x in lc.rows("SELECT number FROM prod_code_series WHERE product_type_code = 'SPARE'")]
fresh = [int(c[3:]) for c in codes if c != "SP-0007"]
lc.check("the new codes are distinct, above the high-water mark (7), and match the claim rows",
len(codes) == 5 and len(fresh) == len(set(fresh)) == 4 and min(fresh) > 7 and sorted(claims) == sorted(fresh), (codes, claims))
# ── dated rows by business date, aware timestamps ─────────────────────────────────────────────
C.mk_product("DATED", name="Dated", base_unit_code="C62", billing_kind="one_off")
C.mk_list("LIST-GBP", rows=[C.row("DATED", "2026-11-10", "10.00"), C.row("DATED", "2026-12-01", "12.00")], default_from=C.DEFAULT)
now, later = must(C.price("DATED", priced_on="2026-11-20")), must(C.price("DATED", priced_on="2026-12-05"))
lc.check("a price is read by business date: 10.00 on 20 Nov, 12.00 on 5 Dec", (now["unit_price"], later["unit_price"]) == ("10.00", "12.00"), (now["unit_price"], later["unit_price"]))
used = api("POST", "/products/BOX/units/PK6/used", SELLING, {"document_ref": "INV-1"})
stamp = lc.rows("SELECT used_at FROM prod_product_unit WHERE owner = 'BOX' AND unit_code = 'PK6'")[0]["used_at"]
print(f" used_at on the wire: {stamp!r}; gtin created_at: {lc.rows('SELECT created_at FROM prod_gtin LIMIT 1')[0]['created_at']!r}")
status, body = api("POST", "/products/BOX/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK6", "contains_unit": "C62", "contains_quantity": "12", "einvoice_unit_code": "XPK"}]})
lc.check("a unit marked used stays fixed (the used_at it reads back, aware or not, is not None): a changed pack size is a 409",
used[0] == 200 and stamp is not None and status == 409, (used, status, str(body)[:120]))
again = api("GET", "/products/BOX", CATALOGUE)
lc.check("the GTINs of the used unit read back in the order they were claimed", again[0] == 200 and len([g for u in again[1]["units"] if u["unit_code"] == "PK6" for g in u["gtins"]]) == 2)
# ── refusal body: mark-unit-used ──────────────────────────────────────────────────────────────
C.mk_product("BOX2", name="Box 2", base_unit_code="C62", billing_kind="one_off", activate=False)
must(api("POST", "/products/BOX2/set-units", CATALOGUE, {"sales_unit_code": "C62", "units_of": [{"unit_code": "PK6", "contains_unit": "C62", "contains_quantity": "6"}]}))
must(api("POST", "/products/BOX2/activate-product", CATALOGUE, {}))
status, body = api("POST", "/products/BOX2/units/PK6/used", SELLING, {"document_ref": "PO-1"})
lc.check("mark-unit-used without an e-invoice code: 409 {gate: einvoice_unit_code, reason, fix}",
status == 409 and body.get("gate") == "einvoice_unit_code" and body.get("reason") and body.get("fix"), (status, body))
# ── GET /product-sites, paged ─────────────────────────────────────────────────────────────────
for i in range(12):
pid = f"SITED-{i:02d}"
C.mk_product(pid, name=pid, base_unit_code="C62", billing_kind="one_off")
must(api("POST", f"/products/{pid}/sites/WH1", PLANNER, {"valid_from": "2026-11-21", "planning_method": "reorder_point", "reorder_point": "10"}))
t0 = time.perf_counter()
listed = must(api("GET", "/product-sites", SALES, {"site_id": "WH1", "on": "2026-11-21", "planning": "active"}))
page1 = must(api("GET", "/product-sites", SALES, {"site_id": "WH1", "on": "2026-11-21", "planning": "active", "limit": 5}))
page2 = must(api("GET", "/product-sites", SALES, {"site_id": "WH1", "on": "2026-11-21", "planning": "active", "limit": 5, "after": page1[-1]["product_id"]}))
before_row = must(api("GET", "/product-sites", SALES, {"site_id": "WH1", "on": "2026-11-20", "planning": "active"}))
print(f" GET /product-sites x4: {(time.perf_counter() - t0) * 1000:.0f} ms")
lc.check("GET /product-sites lists the 12 sited products by business date (none the day before), paged without gaps",
len(listed) == 12 and len(before_row) == 0 and [r["product_id"] for r in page1 + page2] == [r["product_id"] for r in listed][:len(page1) + len(page2)] and len(page1) == 5,
(len(listed), len(before_row), len(page1), len(page2)))
# ── the Odoo migration: chunks of 100, twice, and a stopped run ───────────────────────────────
t0 = time.perf_counter()
first = must(api("POST", "/migrations/odoo/units", IMPORTER, odoo_run("odoo19a", 250)))
t1 = time.perf_counter()
second = must(api("POST", "/migrations/odoo/units", IMPORTER, odoo_run("odoo19a", 250)))
t2 = time.perf_counter()
print(f" migration of 250 products (3 chunks): {(t1 - t0) * 1000:.0f} ms; the repeat: {(t2 - t1) * 1000:.0f} ms")
lc.check("the Odoo migration (250 products, 3 chunks) writes them; run twice, the second writes nothing",
first["written"]["products"] == 250 and first["total_written"] > 250 and second["total_written"] == 0 and n_rows("prod_product", "product_id LIKE 'A-%'") == 250,
(first["written"], second["total_written"]))
real, calls = uc._migration_write, []
def stop_on_second(*a, **kw):
calls.append(1)
if len(calls) == 2:
raise RuntimeError("the run was stopped between chunks")
return real(*a, **kw)
uc._migration_write = stop_on_second
try:
try:
api("POST", "/migrations/odoo/units", IMPORTER, odoo_run("odoo19b", 250))
stopped = False
except RuntimeError:
stopped = True
finally:
uc._migration_write = real
kept = n_rows("prod_product", "product_id LIKE 'B-%'")
resumed = must(api("POST", "/migrations/odoo/units", IMPORTER, odoo_run("odoo19b", 250)))
final = n_rows("prod_product", "product_id LIKE 'B-%'")
done = must(api("POST", "/migrations/odoo/units", IMPORTER, odoo_run("odoo19b", 250)))
lc.check("a run stopped after its first chunk keeps 100 products; run again it writes the other 150 and a third run writes nothing",
stopped and kept == 100 and resumed["written"]["products"] == 150 and final == 250 and done["total_written"] == 0, (stopped, kept, resumed["written"], final, done["total_written"]))
ids = must(api("GET", "/migration-map/odoo19b/product.product", SELLING, {"ids": ",".join(str(100 + i) for i in range(250))}))
lc.check("the migration map answers for every product of the resumed run", len(ids) == 250, len(ids))
# ── the agent door ───────────────────────────────────────────────────────────────────────────
everyone = C.person("all@live", sorted({u.permission for u in uc.PRODUCT.use_cases if u.permission}))
names = C.tools(everyone)
expected = sorted(u.name for u in uc.PRODUCT.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 rec.violations, rec.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}")
_ = Repository
if __name__ == "__main__":
try:
main()
finally: # replaces an exception in flight: finish() records a crash as a failure
raise SystemExit(lc.finish(TABLES))