#!/usr/bin/env python
"""The quote story, 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 policy reads the deepest line as well as the net (H2); the revision on DataK3 — a stale one is
409, and eight lines added at once with no revision are eight lines (the audit kept one); four opens for one deal
at once make one quote (H4); the facts the pipeline reads name each accepted quote's customer (C1).
"""
import pathlib
import sys
import threading
from decimal import Decimal
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_quote.application import use_cases as uc # noqa: E402
from brick_quote.infrastructure.tables import QuoteRepository, md # noqa: E402
rows, check = lc.rows, lc.check
def refused(fn, *a, **kw):
"""(status, reason) if refused — a Refused, or a lost revision (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
out.append(("error", getattr(e, "status", None) or type(e).__name__, str(e)[-160:]))
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 person(email, roles, permissions):
return {"sub": email, "email": email, "roles": roles, "permissions": permissions,
"auth_source": "pool"}
ANNA = person("anna@co.example", ["sales"], ["quote:quote:draft", "quote:quote:decide"])
DESK = person("dee@co.example", ["deal_desk"], ["quote:quote:approve", "quote:discount_policy:write"])
PRICING = person("pat@co.example", ["pricing"], ["quote:price_book:write"])
def main() -> int:
tables = [t.name for t in md.sorted_tables]
lc.create_schema(ROOT / "src" / "brick_quote" / "schema.sql", tables)
uc.repository = lc.repository(QuoteRepository)
for product, price in (("AD-PAGE", "1000"), ("AD-HALF", "600")):
uc.set_price(PRICING, "standard", product, "EUR", "2026-01-01", Decimal(price), product)
uc.set_discount_policy(DESK, Decimal("15"), Decimal("30"))
q = uc.open_quote(ANNA, 4242, "EUR", valid_until="2099-12-31")
qid = q["quote_id"]
first_created = rows(f"SELECT created_at FROM quote_quotes WHERE quote_id = '{qid}'")[0]["created_at"]
uc.add_line(ANNA, qid, 1, "AD-PAGE", 1, Decimal("820"), revenue_months=["2026-11", "2026-12", "2027-01"])
uc.add_line(ANNA, qid, 1, "AD-HALF", 2)
v = uc.get_quote(ANNA, qid)
check("CRM-05 totals computed: 1 × 820 + 2 × 600 = 2020.00, list 2200.00",
(v["total"], v["list_total"]) == (Decimal("2020.00"), Decimal("2200.00")), v)
check("H2 net 8.18%, but the policy reads the deepest line: 18.00%, past the 15% — the desk's",
(v["net_discount_pct"], v["discount_pct"], v["needs_approval"]) == (Decimal("8.18"), Decimal("18.00"), True),
v)
sched = rows(f"SELECT revenue_month, amount FROM quote_line_schedule WHERE quote_id = '{qid}' "
f"ORDER BY revenue_month")
check("schedule stored: 3 months summing to 820.00",
len(sched) == 3 and sum(Decimal(str(r["amount"])) for r in sched) == Decimal("820.00"), sched)
uc.edit_line(ANNA, qid, 1, 1, revenue_months=["2026-11"])
sched = rows(f"SELECT revenue_month FROM quote_line_schedule WHERE quote_id = '{qid}'")
check("months trimmed from a kept line are deleted on DataK3", len(sched) == 1, sched)
past_max = refused(uc.edit_line, ANNA, qid, 1, 2, unit_price=Decimal("300")) # 35.45%
line2 = rows(f"SELECT unit_price FROM quote_lines WHERE quote_id = '{qid}' AND line_no = 2")[0]
check("an edit past the 30% maximum is refused 409 and nothing is stored",
past_max and past_max[0] == 409 and Decimal(str(line2["unit_price"])) == Decimal("600.00"),
(past_max, line2))
uc.edit_line(ANNA, qid, 1, 2, unit_price=Decimal("450")) # (180 + 300) / 2200
v = uc.get_quote(ANNA, qid)
check("now 21.82% net and a 25.00% line — needs the desk at 25.00%",
(v["net_discount_pct"], v["discount_pct"], v["needs_approval"]) == (Decimal("21.82"), Decimal("25.00"), True),
(v["discount_pct"], v["needs_approval"]))
status = refused(uc.send_quote, ANNA, qid, 1)
check("CRM-06 above 15% the send is refused 403 with the figures",
status and status[0] == 403 and str(v["discount_pct"]) in status[1], status)
check("CRM-06 the rep cannot approve (owner)",
refused(uc.approve_discount, person("anna@co.example", ["sales"], ["quote:quote:approve"]),
qid, 1) is not None)
uc.approve_discount(DESK, qid, 1)
stored = rows(f"SELECT approved_by, approved_pct FROM quote_versions WHERE quote_id = '{qid}'")[0]
check("the approval is stored with the figure it covers",
stored["approved_by"] == "dee@co.example"
and Decimal(str(stored["approved_pct"])) == v["discount_pct"], stored)
uc.send_quote(ANNA, qid, 1)
edit_sent = refused(uc.edit_line, ANNA, qid, 1, 1, quantity=5)
check("CRM-07 a sent version refuses an edit", edit_sent and edit_sent[0] == 409, edit_sent)
uc.revise_quote(ANNA, qid)
uc.remove_line(ANNA, qid, 2, 1)
lines_v2 = rows(f"SELECT line_no FROM quote_lines WHERE quote_id = '{qid}' AND version = 2")
sched_v2 = rows(f"SELECT line_no FROM quote_line_schedule WHERE quote_id = '{qid}' AND version = 2")
lines_v1 = rows(f"SELECT line_no FROM quote_lines WHERE quote_id = '{qid}' AND version = 1")
check("v2 lost line 1 and its schedule; v1 still holds both lines as sent",
len(lines_v2) == 1 and sched_v2 == [] and len(lines_v1) == 2, (lines_v2, sched_v2, lines_v1))
v2_send = refused(uc.send_quote, ANNA, qid, 2)
check("a revision does not inherit the approval: v2 at 25.00% needs its own",
v2_send and v2_send[0] == 403 and "25.00%" in v2_send[1], v2_send)
uc.approve_discount(DESK, qid, 2)
uc.send_quote(ANNA, qid, 2)
uc.record_decision(ANNA, qid, 2, "ACCEPTED")
uc.record_decision(ANNA, qid, 2, "ACCEPTED")
statuses = rows(f"SELECT version, status FROM quote_versions WHERE quote_id = '{qid}' ORDER BY version")
check("accepted v2 superseded v1; accepting twice is one acceptance",
[r["status"] for r in statuses] == ["SUPERSEDED", "ACCEPTED"]
and [e["action"] for e in uc.quote_history(ANNA, qid)["events"]].count("accepted") == 1,
statuses)
now_created = rows(f"SELECT created_at, updated_at FROM quote_quotes WHERE quote_id = '{qid}'")[0]
check("created_at survived 14 saves on DataK3 (the tables#417 workaround)",
now_created["created_at"] == first_created and now_created["updated_at"] != first_created,
(first_created, now_created))
in_place = refused(uc.set_price, PRICING, "standard", "AD-PAGE", "EUR", "2026-01-01", Decimal("1100"))
check("a price is never changed in place", in_place and in_place[0] == 409, in_place)
approved = [e["detail"] for e in uc.quote_history(ANNA, qid)["events"] if e["action"] == "discount approved"]
check("history text stored as written", approved == ["v1 at 25.00%", "v2 at 25.00%"], approved)
# ── 0.2.0 ── decided at COMMIT, so through a REAL engine (multi-statement transactions), not the CLI's
# one-statement sessions: there a save's claim, root and lines commit separately and a racer can read half
# of one — a property of the harness, not of DataK3 (measured here: the harness lost 2 of 8 lines).
from contextlib import contextmanager
eng = lc.engine(pool_size=12)
@contextmanager
def real():
with eng.begin() as conn:
yield QuoteRepository(conn)
uc.repository = real
facts = uc.opportunity_quote_facts(person("pipe@svc", ["service"], ["quote:opportunity:read"]), "O-NONE")
check("C1 facts carry each accepted quote's customer (none here)", facts["accepted"] == [], facts)
q2 = uc.open_quote(ANNA, 4242, "EUR", valid_until="2099-12-31")
rev = uc.get_quote(ANNA, q2["quote_id"])["revision"]
uc.add_line(ANNA, q2["quote_id"], 1, "AD-PAGE", 1, revision=rev)
stale = refused(uc.add_line, ANNA, q2["quote_id"], 1, "AD-HALF", 1, revision=rev)
check("C3 a stale revision is 409 on DataK3 and adds nothing",
stale and stale[0] == 409 and len(uc.get_quote(ANNA, q2["quote_id"])["lines"]) == 1, stale)
out = at_once(8, lambda i: uc.add_line(ANNA, q2["quote_id"], 1, "AD-HALF", 1, description=f"line {i}"))
after = uc.get_quote(ANNA, q2["quote_id"])
check("C3 eight lines added at once with no revision: all eight kept, numbered 2..9",
all(k == "ok" for k, *_ in out) and sorted(l["line_no"] for l in after["lines"]) == list(range(1, 10)),
(out, [l["line_no"] for l in after["lines"]]))
opens = at_once(4, lambda i: uc.open_quote(ANNA, 4242, "EUR", opportunity_id="O-LIVE-RACE"))
ids = {r[1]["quote_id"] for r in opens if r[0] == "ok"}
on_deal = rows("SELECT quote_id FROM quote_quotes WHERE opportunity_id = 'O-LIVE-RACE'")
check("H4 four opens for one deal at once on DataK3: one quote; the others answer it",
len(ids) == 1 and len(on_deal) == 1 and all(r[0] == "ok" or r[1] == 409 for r in opens), (opens, on_deal))
return lc.finish(tables)
if __name__ == "__main__":
raise SystemExit(main())