"""The pipeline through both doors over the REAL repository on SQLite — and, in the last section, with
the REAL quote and business-partner bricks running in this process (tier 1), so "won needs an accepted
quote" and "the group's pipeline" are tested across actual bricks, not stand-ins."""
import base64
import json
from contextlib import contextmanager
import anyio
import httpx
import httpx2
import pytest
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from sqlalchemy.pool import StaticPool
from brick_pipeline.app import app
from brick_pipeline.application import use_cases
def sqlite_repo(md, factory):
engine = create_engine("sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False})
md.create_all(engine)
@contextmanager
def repo():
with engine.begin() as conn:
yield factory(conn)
return repo
@pytest.fixture(autouse=True)
def database(monkeypatch):
from brick_pipeline.infrastructure.tables import PipelineRepository, md
monkeypatch.setattr(use_cases, "repository", sqlite_repo(md, PipelineRepository))
def person(email, roles, permissions):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": permissions}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": roles}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
REP = ["pipeline:opportunity:write"]
TOM = person("tom@co.example", ["sales"], REP)
SOFIA = person("sofia@co.example", ["sales"], REP)
MANAGING = ["pipeline:opportunity:read_all", "pipeline:opportunity:manage"] # 0.4.0: permissions, not the role name
BOSS = person("max@co.example", ["manager"], REP + MANAGING + ["pipeline:forecast:override", "pipeline:forecast:snapshot",
"pipeline:stages:write"])
ACTIVITY_APP = person("activity@svc", ["service"], ["pipeline:opportunity:touch"])
ACME = 4242
async def _api(method, path, who, body=None, params=None):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://b") as c:
return await c.request(method, path, headers=who, json=body, params=params)
def api(method, path, who, body=None, params=None):
r = anyio.run(_api, method, path, who, body, params)
return r.status_code, r.json()
async def _tool(who, name, args):
async with app.router.lifespan_context(app):
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=who,
base_url="http://b") as http:
async with streamable_http_client("http://b/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
res = await s.call_tool(name, args)
text = res.content[0].text
return ("refused", text.split(f"Error executing tool {name}: ", 1)[1]) if res.is_error \
else ("ok", json.loads(text))
def deal(who=TOM, amount="10000", close="2026-11-15", bp_id=ACME, name="Acme renewal"):
status, o = api("POST", "/opportunities", who, {"bp_id": bp_id, "name": name, "currency": "EUR",
"amount": amount, "close_date": close})
assert status == 200, o
return o
def move(o, stage, who=TOM, **kw):
return api("POST", f"/opportunities/{o['opportunity_id']}/stage", who, {"stage": stage, **kw})
# ── CRM-01 / CRM-04 ────────────────────────────────────────────────────────────────────────
def test_crm01_a_rep_sees_only_their_own_deals_and_another_reps_is_not_found():
mine, theirs = deal(TOM), deal(SOFIA)
assert [o["opportunity_id"] for o in api("GET", "/opportunities", TOM)[1]["opportunities"]] == \
[mine["opportunity_id"]]
assert api("GET", f"/opportunities/{theirs['opportunity_id']}", TOM)[0] == 404
assert len(api("GET", "/opportunities", BOSS)[1]["opportunities"]) == 2
def test_crm04_moving_a_deal_follows_the_stage_and_the_manager_sees_it_at_once():
o = deal()
status, moved = move(o, "proposal")
assert status == 200 and (moved["probability"], moved["category"], moved["weighted"]) == \
("50.00", "best_case", "5000.00")
assert api("GET", f"/opportunities/{o['opportunity_id']}", BOSS)[1]["stage"] == "proposal"
def test_an_agent_moves_a_deal_through_mcp():
o = deal()
verdict, moved = anyio.run(_tool, TOM, "move_opportunity",
{"opportunity_id": o["opportunity_id"], "stage": "negotiation"})
assert verdict == "ok" and moved["category"] == "commit"
def test_lost_needs_a_listed_reason_through_both_doors():
o = deal()
status, body = move(o, "closed_lost", loss_reason="went quiet")
assert status == 422 and "price, no_budget" in body["detail"]
verdict, said = anyio.run(_tool, TOM, "move_opportunity", {"opportunity_id": o["opportunity_id"],
"stage": "closed_lost", "loss_reason": "went quiet"})
assert said == f"422: {body['detail']}"
assert move(o, "closed_lost", loss_reason="competitor")[1]["loss_reason"] == "competitor"
def test_without_a_quote_module_won_is_recorded_and_says_so():
o = deal()
assert move(o, "closed_won")[1]["stage"] == "closed_won"
events = api("GET", f"/opportunities/{o['opportunity_id']}/history", TOM)[1]["events"]
assert "won without an accepted-quote check" in events[-1]["detail"]
def test_another_rep_cannot_move_my_deal_but_a_manager_can():
o = deal(TOM)
assert move(o, "proposal", who=SOFIA)[0] == 404
assert move(o, "proposal", who=BOSS)[0] == 200
# ── CRM-09 / CRM-15 ────────────────────────────────────────────────────────────────────────
def test_crm09_the_forecast_and_a_reasoned_override():
a, b = deal(TOM, "10000"), deal(TOM, "20000", "2026-12-01")
move(a, "proposal")
move(b, "proposal")
status, body = api("PUT", f"/opportunities/{b['opportunity_id']}/category", BOSS, {"category": "commit"})
assert status == 422
api("PUT", f"/opportunities/{b['opportunity_id']}/category", BOSS,
{"category": "commit", "reason": "verbal yes from the CFO"})
row = api("GET", "/forecast", BOSS, params={"period": "2026-Q4"})[1]["rows"][0]
assert (row["open"], row["weighted"], row["commit"]) == ("30000.00", "15000.00", "20000.00")
assert api("PUT", f"/opportunities/{a['opportunity_id']}/category", TOM,
{"category": "commit", "reason": "x"})[0] == 403
def test_crm15_a_snapshot_rerun_replaces_it_and_drops_rows_that_no_longer_apply():
a, b = deal(TOM), deal(SOFIA)
api("POST", "/forecast/snapshots", BOSS, {"snapshot_date": "2026-11-01"})
assert {r["owner"] for r in api("GET", "/forecast/snapshots/2026-11-01", BOSS)[1]["rows"]} == \
{"tom@co.example", "sofia@co.example"}
api("PATCH", f"/opportunities/{b['opportunity_id']}", BOSS, {"close_date": "2027-02-01"})
move(b, "closed_lost", who=SOFIA, loss_reason="price")
api("POST", "/forecast/snapshots", BOSS, {"snapshot_date": "2026-11-01"})
rows = api("GET", "/forecast/snapshots/2026-11-01", BOSS)[1]["rows"]
assert [(r["owner"], r["period"]) for r in rows] == [("tom@co.example", "2026-Q4"),
("sofia@co.example", "2027-Q1")]
assert [r["owner"] for r in api("GET", "/forecast/snapshots/2026-11-01", TOM)[1]["rows"]] == \
["tom@co.example"]
# ── CRM-16 and the activity seam ───────────────────────────────────────────────────────────
def test_crm16_a_quiet_late_deal_is_at_risk_until_activity_is_reported(monkeypatch):
o = deal(close="2100-06-01") # after the faked "today", so only quietness can flag it
move(o, "negotiation")
import brick_pipeline.domain.pipeline as dom
real_risk = dom.Opportunity.risk
monkeypatch.setattr(dom.Opportunity, "risk",
lambda self, p, policy, today=None: real_risk(self, p, policy, today="2099-01-01"))
at_risk = api("GET", "/opportunities", TOM, params={"at_risk": True})[1]["opportunities"]
assert [x["opportunity_id"] for x in at_risk] == [o["opportunity_id"]]
assert "nothing logged" in at_risk[0]["at_risk"]
assert api("POST", f"/opportunities/{o['opportunity_id']}/touch", ACTIVITY_APP,
{"at": "2098-12-30T10:00:00Z", "what": "call with the CFO"})[0] == 200
assert api("GET", "/opportunities", TOM, params={"at_risk": True})[1]["opportunities"] == []
assert api("POST", f"/opportunities/{o['opportunity_id']}/touch", TOM, {"at": "2099-01-01"})[0] == 403
# ── 0.2.0: one source document, one deal — 0.3.0: from a machine only (C2) ────────────────
LEAD_APP = person("tom@co.example", ["service"], ["pipeline:opportunity:write", "pipeline:opportunity:link_source"])
def _open_from(ref, who=LEAD_APP, name="Acme from lead", bp_id=ACME, owner="tom@co.example"):
return api("POST", "/opportunities/from-source", who, {"bp_id": bp_id, "name": name, "currency": "EUR",
"source_ref": ref, "owner": owner})
def test_crm03_the_same_source_ref_always_returns_the_same_deal():
status, first = _open_from("lead:L-1")
assert status == 200 and first["owner"] == "tom@co.example" and "from lead:L-1" in api(
"GET", f"/opportunities/{first['opportunity_id']}/history", TOM)[1]["events"][0]["detail"]
assert _open_from("lead:L-1", name="a retry with another name")[1]["opportunity_id"] == first["opportunity_id"]
assert [o["opportunity_id"] for o in api("GET", "/opportunities", TOM)[1]["opportunities"]] == \
[first["opportunity_id"]]
assert _open_from("lead:L-2")[1]["opportunity_id"] != first["opportunity_id"]
deal(name="no ref")
deal(name="no ref") # without a source_ref every call is a new deal, as before
assert len(api("GET", "/opportunities", TOM)[1]["opportunities"]) == 4
def test_c2_a_person_cannot_name_a_source_ref():
"""Tom planted a deal under the ref of Sofia's lead, and her conversion then landed on it (the CRM audit,
2026-10-17). source_ref is gone from the person's route and tool; the machine route needs a permission no
person role holds, and it is no agent's tool."""
planted = api("POST", "/opportunities", TOM, {"bp_id": 1, "name": "planted", "currency": "EUR",
"source_ref": "lead:L-sofia"})[1]
assert _open_from("lead:L-sofia", owner="sofia@co.example")[1]["opportunity_id"] != planted["opportunity_id"]
assert api("POST", "/opportunities/from-source", TOM, {"bp_id": ACME, "name": "x", "currency": "EUR",
"source_ref": "lead:L-x", "owner": "tom@co.example"}) == \
(403, {"detail": "missing permission: pipeline:opportunity:link_source"})
tools = anyio.run(_tools, TOM)
assert "open_opportunity_from_source" not in tools
assert "source_ref" not in tools["open_opportunity"].input_schema["properties"]
verdict, via_mcp = anyio.run(_tool, TOM, "open_opportunity", {"bp_id": ACME, "name": "x", "currency": "EUR",
"source_ref": "lead:L-y"})
assert _open_from("lead:L-y")[1]["opportunity_id"] != via_mcp["opportunity_id"]
async def _tools(who):
async with app.router.lifespan_context(app):
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=who,
base_url="http://b") as http:
async with streamable_http_client("http://b/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
return {t.name: t for t in (await s.list_tools()).tools}
def test_c2_a_reused_ref_must_be_for_the_same_partner_and_writes_nothing_when_not():
first = _open_from("lead:L-7", bp_id=ACME)[1]
status, body = _open_from("lead:L-7", bp_id=999, name="other partner")
assert status == 409 and "partner" in body["detail"]
assert [o["name"] for o in api("GET", "/opportunities", BOSS)[1]["opportunities"]] == ["Acme from lead"]
assert api("GET", f"/opportunities/{first['opportunity_id']}", BOSS)[1]["bp_id"] == str(ACME)
def test_a_raced_source_ref_returns_the_winner_and_writes_nothing_of_its_own(monkeypatch):
"""Two conversions of one lead, interleaved: the loser looked the ref up BEFORE the winner committed (so
it saw nothing), then tried to claim it AFTER (so its plain INSERT collides). SQLite has one connection
here, so the interleaving is staged by making the loser's lookup miss — the collision, the rollback and
the re-read are the real code on a real table."""
from brick_pipeline.infrastructure.tables import PipelineRepository
winner = _open_from("lead:L-race")[1]
real, calls = PipelineRepository.by_source, []
def lookup_before_the_winner_committed(self, ref):
calls.append(ref)
return None if len(calls) == 1 else real(self, ref)
monkeypatch.setattr(PipelineRepository, "by_source", lookup_before_the_winner_committed)
status, loser = _open_from("lead:L-race", name="the loser")
assert status == 200 and loser["opportunity_id"] == winner["opportunity_id"]
assert len(calls) == 2 # it did collide, and re-read
deals = api("GET", "/opportunities", BOSS)[1]["opportunities"]
assert [o["name"] for o in deals] == ["Acme from lead"] # the loser's deal was rolled back with its claim
# ── 0.3.0: the CRM audit's wave 1 (2026-10-17) ─────────────────────────────────────────────
REOPENER = person("max@co.example", ["manager"], REP + MANAGING + ["pipeline:opportunity:reopen"])
def tool(who, name, args):
return anyio.run(_tool, who, name, args)
def test_h3_a_closed_deal_is_final_and_only_a_manager_reopens_it_with_a_reason():
o = deal()
move(o, "closed_lost", loss_reason="price")
status, body = move(o, "discovery")
assert status == 409 and "reopen" in body["detail"]
assert api("PATCH", f"/opportunities/{o['opportunity_id']}", TOM, {"close_date": "2027-01-01"})[0] == 409
assert api("PATCH", f"/opportunities/{o['opportunity_id']}", TOM, {"amount": "1"})[0] == 409
path = f"/opportunities/{o['opportunity_id']}/reopen"
assert api("POST", path, TOM, {"reason": "they called back"}) == \
(403, {"detail": "missing permission: pipeline:opportunity:reopen"})
assert api("POST", path, REOPENER, {"reason": " "})[0] == 422
status, back = api("POST", path, REOPENER, {"reason": "they called back", "stage": "negotiation"})
assert status == 200 and back["stage"] == "negotiation" and back["loss_reason"] is None
events = api("GET", f"/opportunities/{o['opportunity_id']}/history", TOM)[1]["events"]
assert events[-1]["action"] == "reopened" and "they called back" in events[-1]["detail"]
assert tool(REOPENER, "reopen_opportunity", {"opportunity_id": o["opportunity_id"], "reason": "x"})[1] \
.startswith("409") # it is open now
def test_c3_a_deal_carries_its_version_and_a_stale_one_is_409():
o = deal()
assert o["version"] == 1
assert api("PATCH", f"/opportunities/{o['opportunity_id']}", TOM, {"name": "v2", "version": 1})[1]["version"] == 2
stale = api("PATCH", f"/opportunities/{o['opportunity_id']}", BOSS, {"amount": "5", "version": 1})
assert stale == (409, {"detail": "changed since you opened it — reload it and make your change again"})
assert move(o, "proposal", version=1)[0] == 409
assert tool(TOM, "move_opportunity", {"opportunity_id": o["opportunity_id"], "stage": "proposal",
"version": 1})[1].startswith("409: changed since you opened it")
r = anyio.run(_raw, "GET", f"/opportunities/{o['opportunity_id']}", TOM, None, {})
assert r.headers["etag"] == '"2"'
r = anyio.run(_raw, "POST", f"/opportunities/{o['opportunity_id']}/stage", TOM, {"stage": "proposal"},
{"If-Match": '"2"'})
assert r.status_code == 200 and r.json()["version"] == 3
def test_c3_two_people_editing_one_deal_without_a_version_both_land(monkeypatch):
"""The audit's race (6 of 6 lost): Tom renames while the manager re-prices. The second request read the
deal before the first committed — what DataK3's snapshot hands it — so it is re-run on the latest."""
from brick_pipeline.infrastructure.tables import PipelineRepository
o = deal()
with use_cases.repository() as r:
snapshot = r.get(o["opportunity_id"])
api("PATCH", f"/opportunities/{o['opportunity_id']}", TOM, {"name": "renamed by Tom"})
real, once = PipelineRepository.get, {"n": 1}
def stale(self, oid):
if once["n"] and oid == o["opportunity_id"]:
once["n"] -= 1
import copy
return copy.deepcopy(snapshot)
return real(self, oid)
monkeypatch.setattr(PipelineRepository, "get", stale)
assert api("PATCH", f"/opportunities/{o['opportunity_id']}", BOSS, {"amount": "1005"})[0] == 200
now = api("GET", f"/opportunities/{o['opportunity_id']}", TOM)[1]
assert (now["name"], now["amount"], now["version"]) == ("renamed by Tom", "1005.00", 3)
async def _raw(method, path, who, body, headers):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://b") as c:
return await c.request(method, path, headers={**who, **headers}, json=body)
def test_c4_ids_leave_as_strings_and_come_in_as_either():
o = deal(bp_id="45665172796086509")
assert o["bp_id"] == "45665172796086509"
assert tool(TOM, "open_opportunity", {"bp_id": 45665172796086509, "name": "n", "currency": "EUR"})[1][
"bp_id"] == "45665172796086509"
group = api("GET", "/partners/45665172796086509/group-pipeline", TOM)[1]
assert group["group"] == ["45665172796086509"] and group["bp_id"] == "45665172796086509"
def test_h12_a_bad_input_is_422_never_500():
for body in ({"bp_id": ACME, "name": "x", "currency": "EUR", "amount": 1e20},
{"bp_id": ACME, "name": "x", "currency": "EUR", "amount": "99999999999999999.99"},
{"bp_id": ACME, "name": "x", "currency": "EUR", "close_date": "2026-13-01"},
{"bp_id": "99999999999999999999999", "name": "x", "currency": "EUR"},
{"bp_id": ACME, "name": "N" * 300, "currency": "EUR"},
{"bp_id": ACME, "name": "nul\x00", "currency": "EUR"}):
status, out = api("POST", "/opportunities", TOM, body)
assert status == 422, (body, status, out)
assert tool(TOM, "open_opportunity", body)[0] == "refused"
assert api("GET", "/forecast/snapshots/not-a-date", BOSS)[0] == 422
assert api("POST", "/forecast/snapshots", BOSS, {"snapshot_date": "not-a-date"})[0] == 422
o = deal()
assert api("PATCH", f"/opportunities/{o['opportunity_id']}", TOM, {"probability": 150})[0] == 422
def test_h13_a_retried_open_with_one_idempotency_key_is_one_deal(tmp_path, monkeypatch):
from erp_platform import idempotency
e = create_engine(f"sqlite:///{tmp_path / 'idem.db'}")
idempotency.md.create_all(e)
monkeypatch.setattr(use_cases.PIPE, "idempotency_connection", e.begin)
body = {"bp_id": ACME, "name": "Acme", "currency": "EUR"}
a = anyio.run(_raw, "POST", "/opportunities", TOM, body, {"Idempotency-Key": "open-1"})
b = anyio.run(_raw, "POST", "/opportunities", TOM, body, {"Idempotency-Key": "open-1"})
assert a.json() == b.json() and b.headers["idempotent-replayed"] == "true"
assert len(api("GET", "/opportunities", TOM)[1]["opportunities"]) == 1
# ── configuration ──────────────────────────────────────────────────────────────────────────
def test_the_stages_are_the_customers_and_a_stage_with_deals_cannot_vanish():
o = deal()
move(o, "qualification")
stages = [{"name": "lead", "position": 1, "probability": 10, "category": "pipeline"},
{"name": "won", "position": 9, "probability": 100, "category": "closed", "is_won": True,
"is_closed": True},
{"name": "lost", "position": 10, "probability": 0, "category": "omitted", "is_closed": True}]
status, body = api("PUT", "/pipeline", BOSS, {"stages": stages})
assert status == 409 and "qualification still holds 1" in body["detail"]
stages.insert(1, {"name": "qualification", "position": 2, "probability": 30, "category": "pipeline"})
assert api("PUT", "/pipeline", BOSS, {"stages": stages})[0] == 200
assert api("GET", f"/opportunities/{o['opportunity_id']}", TOM)[1]["probability"] == "30.00"
assert api("PUT", "/pipeline", TOM, {"stages": stages})[0] == 403
# ── tier 1: the real quote and business-partner bricks in this process ─────────────────────
# (and this pipeline registered too, so quote's open_quote can look the deal up as the caller — C1)
@pytest.fixture
def suite(monkeypatch):
pytest.importorskip("brick_quote")
pytest.importorskip("brick_bp")
from erp_platform.seam import register_local, unregister_local
from brick_bp.app import app as bp_app
from brick_bp.application import use_cases as bp_uc
from brick_bp.infrastructure.tables import PartnerRepository, md as bp_md
from brick_quote.app import app as quote_app
from brick_quote.application import use_cases as q_uc
from brick_quote.infrastructure.tables import QuoteRepository, md as q_md
monkeypatch.setattr(bp_uc, "repository", sqlite_repo(bp_md, PartnerRepository))
monkeypatch.setattr(q_uc, "repository", sqlite_repo(q_md, QuoteRepository))
steward = {"sub": "s", "email": "s@co.example", "roles": [], "auth_source": "pool",
"permissions": ["bp:partner:write", "bp:family:write"]}
ids = {k: bp_uc.upsert_partner(steward, k, n, c, roles=["customer"])["bp_id"]
for k, n, c in (("acme-group", "Acme Group", "GB"), ("acme-uk", "Acme UK", "GB"),
("acme-de", "Acme DE", "DE"), ("other", "Other Co", "FR"))}
bp_uc.link_partners(steward, ids["acme-uk"], ids["acme-group"])
bp_uc.link_partners(steward, ids["acme-de"], ids["acme-group"])
with q_uc.repository() as r:
r.set_price("standard", "AD-PAGE", "EUR", "2026-01-01", "1000", "AD-PAGE", "seed")
for name, brick in (("business-partner", bp_app), ("quote", quote_app), ("pipeline", app)):
register_local(name, brick)
rep = {"sub": "tom@co.example", "email": "tom@co.example", "roles": ["sales"], "auth_source": "pool",
"permissions": ["quote:quote:draft", "quote:quote:decide"]}
yield {"ids": ids, "q": q_uc, "rep": rep, "block": lambda k: bp_uc.set_partner_status(
steward, ids[k], "blocked", "unpaid")}
for name in ("business-partner", "quote", "pipeline"):
unregister_local(name)
def test_won_needs_the_deal_to_have_an_accepted_quote_and_takes_its_total(suite):
o = api("POST", "/opportunities", TOM, {"bp_id": suite["ids"]["acme-uk"], "name": "UK print",
"currency": "EUR", "close_date": "2026-12-01"})[1]
status, body = move(o, "closed_won")
assert status == 422 and "accepted a quote" in body["detail"]
accept(suite, suite["ids"]["acme-uk"], o, 3)
status, won = move(o, "closed_won")
assert status == 200 and won["stage"] == "closed_won" and won["amount"] == "3000.00"
def accept(suite, bp_id, o, quantity, *, check_deal=True):
"""A quote for `bp_id` on deal `o`, sent and accepted. check_deal=False opens it the way 0.1 could: with
the pipeline not asked (a quote attached to any deal id), which is what C1 closes."""
from erp_platform.seam import register_local, unregister_local
q, rep = suite["q"], suite["rep"]
if not check_deal:
unregister_local("pipeline")
try:
quote = q.open_quote(rep, bp_id, "EUR", opportunity_id=o["opportunity_id"], valid_until="2099-12-31")
finally:
register_local("pipeline", app)
q.add_line(rep, quote["quote_id"], 1, "AD-PAGE", quantity)
q.send_quote(rep, quote["quote_id"], 1)
q.record_decision(rep, quote["quote_id"], 1, "ACCEPTED")
return quote
def test_c1_an_accepted_quote_for_another_customer_does_not_win_the_deal(suite):
o = deal(bp_id=suite["ids"]["acme-uk"], amount="10000")
accept(suite, suite["ids"]["other"], o, 1, check_deal=False) # Other Co's quote, on Acme UK's deal
status, body = move(o, "closed_won")
assert status == 422 and "accepted a quote" in body["detail"]
accept(suite, suite["ids"]["acme-uk"], o, 2)
status, won = move(o, "closed_won")
assert status == 200 and won["amount"] == "2000.00" # H1: Acme UK's accepted total
def test_h1_h3_won_takes_the_quote_and_reopening_voids_it(suite):
o = deal(bp_id=suite["ids"]["acme-uk"], amount="10000")
q1 = accept(suite, suite["ids"]["acme-uk"], o, 3)
assert move(o, "closed_won")[1]["amount"] == "3000.00" # not the 10 000 typed
path = f"/opportunities/{o['opportunity_id']}/reopen"
assert api("POST", path, REOPENER, {"reason": "the order was cancelled"})[0] == 200
status, body = move(o, "closed_won")
assert status == 422 and "accepted a quote" in body["detail"] # Q1's win was voided
accept(suite, suite["ids"]["acme-uk"], o, 4)
status, won = move(o, "closed_won")
assert status == 200 and won["amount"] == "4000.00"
events = api("GET", f"/opportunities/{o['opportunity_id']}/history", BOSS)[1]["events"]
assert any(q1["quote_id"] in e["detail"] for e in events if e["action"] == "reopened")
def test_crm08_the_group_pipeline_spans_the_corporate_family(suite):
ids = suite["ids"]
uk = api("POST", "/opportunities", TOM, {"bp_id": ids["acme-uk"], "name": "UK", "currency": "EUR"})[1]
de = api("POST", "/opportunities", SOFIA, {"bp_id": ids["acme-de"], "name": "DE", "currency": "EUR"})[1]
api("POST", "/opportunities", TOM, {"bp_id": ids["other"], "name": "elsewhere", "currency": "EUR"})
group = api("GET", f"/partners/{ids['acme-uk']}/group-pipeline", BOSS)[1]
assert {o["opportunity_id"] for o in group["opportunities"]} == {uk["opportunity_id"], de["opportunity_id"]}
assert set(group["group"]) == {str(ids["acme-group"]), str(ids["acme-uk"]), str(ids["acme-de"])}
rep_view = api("GET", f"/partners/{ids['acme-uk']}/group-pipeline", TOM)[1]
assert [o["opportunity_id"] for o in rep_view["opportunities"]] == [uk["opportunity_id"]]
def test_no_new_deal_for_a_blocked_customer_or_an_unknown_one(suite):
suite["block"]("other")
status, body = api("POST", "/opportunities", TOM, {"bp_id": suite["ids"]["other"], "name": "x",
"currency": "EUR"})
assert status == 409 and "Other Co is blocked" in body["detail"]
assert api("POST", "/opportunities", TOM, {"bp_id": 1, "name": "x", "currency": "EUR"})[0] == 422
# ── 0.3.1: link_source is a MACHINE permission — no wildcard reaches it (erp-platform 0.9.2) ──────────
def test_a_wildcard_grant_never_reaches_the_machine_route():
"""`*:*:*` (and `pipeline:*:*`, `pipeline:opportunity:*`) runs every person command and not this one: naming the
source a deal opens from is how a conversion finds its deal again, so only an EXACT grant — a service client's —
passes. The wildcard holder still runs the person commands."""
body = {"bp_id": ACME, "name": "x", "currency": "EUR", "source_ref": "lead:L-w", "owner": "tom@co.example"}
for grant in ("*:*:*", "pipeline:*:*", "pipeline:opportunity:*"):
star = person("star@co.example", ["anything"], [grant])
assert api("POST", "/opportunities/from-source", star, body) == \
(403, {"detail": "missing permission: pipeline:opportunity:link_source"}), grant
assert api("POST", "/opportunities", star, {"bp_id": ACME, "name": "y", "currency": "EUR"})[0] == 200, grant
assert _open_from("lead:L-w")[0] == 200 # the exact grant passes
def test_the_process_is_served_as_bpmn():
status, view = api("GET", "/process/bpmn", TOM)
assert status == 200, view
text = json.dumps(view)
for key in ("open_opportunity", "move_stage", "win", "lose", "reopen_opportunity"):
assert f"Task_{key}" in text, key
# ── 0.4.0: a brick decides by PERMISSION, never by role name (erp-platform 0.9.2) ─────────────────────────
def test_the_default_roles_name_every_permission_this_brick_decides_by():
"""src/brick_pipeline/roles.yaml — the roles this brick ships, from its actors. They hold exactly the permissions the brick
decides by (its use cases' and READ_ALL, MANAGE), and a machine permission only on a service client."""
import pathlib
import yaml
from brick_pipeline.application.use_cases import PIPE, READ_ALL, MANAGE
spec = yaml.safe_load((pathlib.Path(use_cases.__file__).parents[1] / "roles.yaml").read_text())
known = {uc.permission for uc in PIPE.use_cases if uc.permission} | {READ_ALL, MANAGE}
assert {p for grants in spec["roles"].values() for p in grants} == known
for role, grants in spec["roles"].items():
assert role in spec.get("service_clients", []) or not set(grants) & set(PIPE.machine_permissions), role
def test_a_role_name_decides_nothing():
"""The old role names (`manager`, `analyst`) without the permissions see and change only their own; any role
holding pipeline:opportunity:read_all sees every deal, and only pipeline:opportunity:manage changes another's."""
toms, sofias = deal(TOM), deal(SOFIA, name="Sofia's")
named = person("nam@co.example", ["manager", "analyst"], REP)
assert api("GET", "/opportunities", named)[1]["opportunities"] == []
assert api("GET", f"/opportunities/{toms['opportunity_id']}", named)[0] == 404
assert api("GET", "/forecast", named)[1]["rows"] == []
reader = person("ops@co.example", ["revops"], REP + ["pipeline:opportunity:read_all"])
assert {o["opportunity_id"] for o in api("GET", "/opportunities", reader)[1]["opportunities"]} == \
{toms["opportunity_id"], sofias["opportunity_id"]}
assert api("PATCH", f"/opportunities/{toms['opportunity_id']}", reader, {"name": "x"})[0] == 403
changer = person("lead@co.example", ["team_lead"], REP + MANAGING)
assert api("PATCH", f"/opportunities/{toms['opportunity_id']}", changer, {"name": "x"})[1]["name"] == "x"