#!/usr/bin/env python
"""The pipeline stories, end to end, on a real DataK3 bucket (erp_platform.livecheck).
BUCKET=<scratch bucket> DODIL_ORG=<org> python scripts/live_check.py
0.3.0 adds: source_ref only through the machine route (open_opportunity_from_source) and a reused ref for another
partner refused; a closed deal is final until a manager reopens it with a reason; the optimistic version on
DataK3 — a stale one is 409 and four edits racing from one version give one success.
"""
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_pipeline.application import use_cases as uc # noqa: E402
from brick_pipeline.infrastructure.tables import PipelineRepository, SourceRefTaken, md # noqa: E402
BIG = 54848654143250435 # a real-shaped bp_id: above 2^53
def person(email, roles, permissions):
return {"sub": email, "email": email, "roles": roles, "permissions": permissions, "auth_source": "pool"}
TOM = person("tom@co.example", ["sales"], ["pipeline:opportunity:write"])
SOFIA = person("sofia@co.example", ["sales"], ["pipeline:opportunity:write"])
BOSS = person("max@co.example", ["manager"], ["pipeline:opportunity:write", "pipeline:forecast:override",
"pipeline:forecast:snapshot", "pipeline:stages:write",
"pipeline:opportunity:reopen"])
LEAD_APP = person("tom@co.example", ["service"], ["pipeline:opportunity:write", "pipeline:opportunity:link_source"])
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 main() -> int:
tables = [t.name for t in md.sorted_tables]
lc.create_schema(ROOT / "src" / "brick_pipeline" / "schema.sql", tables)
uc.repository = lc.repository(PipelineRepository)
a = uc.open_opportunity(TOM, BIG, "Acme renewal", "EUR", Decimal("10000"), "2026-11-15")
b = uc.open_opportunity(TOM, BIG, "Acme print", "EUR", Decimal("20000"), "2026-12-01")
s = uc.open_opportunity(SOFIA, 4242, "Other deal", "EUR", Decimal("8000"), "2026-11-20")
lc.check("bp_id above 2^53 round-trips exactly", uc.get_opportunity(TOM, a["opportunity_id"])["bp_id"] == BIG)
uc.move_opportunity(TOM, a["opportunity_id"], "proposal")
moved = uc.move_opportunity(TOM, b["opportunity_id"], "negotiation")
lc.check("CRM-04 the stage sets probability, category and weighted value",
(moved["probability"], moved["category"], moved["weighted"]) ==
(Decimal("75.00"), "commit", Decimal("15000.00")), moved)
lost = lc.refused(uc.move_opportunity, SOFIA, s["opportunity_id"], "closed_lost", "went quiet")
lc.check("lost needs a listed reason", lost and lost[0] == 422, lost)
lc.check("CRM-01 a rep cannot see another rep's deal",
(lc.refused(uc.get_opportunity, SOFIA, a["opportunity_id"]) or (0,))[0] == 404)
uc.override_category(BOSS, a["opportunity_id"], "commit", "verbal yes")
tom = [r for r in uc.get_forecast(BOSS, "2026-Q4")["rows"] if r["owner"] == "tom@co.example"][0]
lc.check("CRM-09 forecast: open 30000, weighted 20000, commit 30000",
(tom["open"], tom["weighted"], tom["commit"]) ==
(Decimal("30000.00"), Decimal("20000.00"), Decimal("30000.00")), tom)
uc.take_forecast_snapshot(BOSS, "2026-11-01")
first = uc.get_forecast_snapshot(BOSS, "2026-11-01")["rows"]
uc.edit_opportunity(BOSS, s["opportunity_id"], close_date="2027-02-01")
uc.move_opportunity(SOFIA, s["opportunity_id"], "closed_lost", "price")
uc.take_forecast_snapshot(BOSS, "2026-11-01")
second = uc.get_forecast_snapshot(BOSS, "2026-11-01")["rows"]
stored = lc.rows("SELECT owner, period FROM pipeline_snapshot_rows ORDER BY period, owner")
lc.check("CRM-15 a re-run replaces the date: Sofia's 2026-Q4 row is gone, her 2027-Q1 row is there",
{(r["owner"], r["period"]) for r in first} == {("tom@co.example", "2026-Q4"),
("sofia@co.example", "2026-Q4")}
and [(r["owner"], r["period"]) for r in stored] == [("tom@co.example", "2026-Q4"),
("sofia@co.example", "2027-Q1")], stored)
uc.touch_opportunity(person("act@svc", ["service"], ["pipeline:opportunity:touch"]),
a["opportunity_id"], "2026-10-16T08:00:00Z", "call")
lc.check("an older activity report does not move last_activity_at backwards",
uc.get_opportunity(TOM, a["opportunity_id"])["last_activity_at"] > "2026-10-16T08:00:00")
stages = [{"name": "discovery", "position": 1, "probability": 10, "category": "pipeline"},
{"name": "proposal", "position": 2, "probability": 40, "category": "best_case"},
{"name": "negotiation", "position": 3, "probability": 80, "category": "commit"},
{"name": "won", "position": 9, "probability": 100, "category": "closed", "is_won": True,
"is_closed": True},
{"name": "closed_lost", "position": 10, "probability": 0, "category": "omitted", "is_closed": True}]
uc.set_pipeline(BOSS, stages)
lc.check("the customer's stages are rows: proposal is now 40%",
uc.get_opportunity(TOM, a["opportunity_id"])["probability"] == Decimal("40.00"))
stages.pop(2)
gone = lc.refused(uc.set_pipeline, BOSS, stages)
lc.check("a stage that holds a deal cannot be removed", gone and gone[0] == 409, gone)
# 0.2.0: one source document, one deal — the claim is a plain INSERT, and DataK3 must refuse the second.
first = uc.open_opportunity_from_source(LEAD_APP, "lead:L-LIVE", BIG, "From lead", "EUR", "tom@co.example")
again = uc.open_opportunity_from_source(LEAD_APP, "lead:L-LIVE", BIG, "From lead (retry)", "EUR",
"tom@co.example")
lc.check("source_ref: a repeat returns the same deal", again["opportunity_id"] == first["opportunity_id"],
(first["opportunity_id"], again["opportunity_id"]))
try:
with uc.repository() as repo:
repo.claim_source("lead:L-LIVE", "O-INTRUDER")
collided = False
except SourceRefTaken:
collided = True
stored = lc.rows("SELECT opportunity_id FROM pipeline_source_refs WHERE source_ref = 'lead:L-LIVE'")
lc.check("source_ref: a second claim fails with 23505 and the first claim stands",
collided and stored == [{"opportunity_id": first["opportunity_id"]}], (collided, stored))
lc.check("source_ref: one deal named 'From lead'",
sum(1 for o in uc.list_opportunities(BOSS, open_only=False)["opportunities"]
if o["name"].startswith("From lead")) == 1)
other = refused(uc.open_opportunity_from_source, LEAD_APP, "lead:L-LIVE", 4242, "Elsewhere", "EUR",
"tom@co.example")
lc.check("C2 a reused ref for another partner is 409, and no deal is written",
other and other[0] == 409 and sum(1 for o in uc.list_opportunities(BOSS, open_only=False)["opportunities"]
if o["name"] == "Elsewhere") == 0, other)
hist = [e["action"] for e in uc.opportunity_history(TOM, a["opportunity_id"])["events"]]
lc.check("history: the stale activity report was not recorded", hist == ["opened", "moved", "category"], hist)
# ── 0.3.0 ───────────────────────────────────────────────────────────────────────────────
lost = uc.get_opportunity(SOFIA, s["opportunity_id"])
final = refused(uc.move_opportunity, SOFIA, s["opportunity_id"], "discovery")
lc.check("H3 a lost deal is final: moving it is 409", final and final[0] == 409 and "reopen" in final[1], final)
back = uc.reopen_opportunity(BOSS, s["opportunity_id"], "they called back", "proposal", version=lost["version"])
lc.check("H3 a manager reopens it with a reason", back["stage"] == "proposal" and back["loss_reason"] is None, back)
v = uc.get_opportunity(TOM, a["opportunity_id"])["version"]
uc.edit_opportunity(TOM, a["opportunity_id"], name="Acme renewal v2", version=v)
stale = refused(uc.edit_opportunity, BOSS, a["opportunity_id"], amount=Decimal("1"), version=v)
lc.check("C3 a stale version is 409 on DataK3 and writes nothing",
stale and stale[0] == 409 and uc.get_opportunity(TOM, a["opportunity_id"])["amount"] == Decimal("10000.00"),
stale)
v = uc.get_opportunity(TOM, a["opportunity_id"])["version"]
out = []
def race(i):
r = refused(uc.edit_opportunity, TOM, a["opportunity_id"], name=f"race {i}", version=v)
out.append(r[0] if r else "ok")
ts = [threading.Thread(target=race, args=(i,)) for i in range(4)]
for t in ts:
t.start()
for t in ts:
t.join()
lc.check("C3 four edits racing from one version: one success, the rest 409",
out.count("ok") == 1 and out.count(409) == 3, out)
return lc.finish(tables)
if __name__ == "__main__":
raise SystemExit(main())