dodil/erp-brick-quote
tests/test_doors.py
"""The use cases through both doors — the API and MCP — over the REAL repository on SQLite.
Everything is real except the database engine: routes, /mcp, permissions, row visibility, the
row mapping and the aggregate save. SQLite runs the same INSERT … ON CONFLICT the repository sends.
It does NOT reproduce DataK3's differences (tables#417), which is what scripts/live_check.py is for.
"""
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_quote.app import app
from brick_quote.application import use_cases
from brick_quote.infrastructure.tables import QuoteRepository, md
@pytest.fixture(autouse=True)
def database(monkeypatch):
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 QuoteRepository(conn)
monkeypatch.setattr(use_cases, "repository", repo)
with repo() as r: # the price book the stories assume
for product, price in (("AD-PAGE", "1000"), ("AD-HALF", "600"), ("DIGITAL", "250")):
r.set_price("standard", product, "EUR", "2026-01-01", price, product, "seed")
return engine
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"}
ACME = 4242 # a bp_id; no partner brick is installed in most tests
REP = ["quote:quote:draft", "quote:quote:decide"]
ANNA = person("anna@co.example", ["sales"], REP)
BEN = person("ben@co.example", ["sales"], REP)
DESK = person("dee@co.example", ["deal_desk"], ["quote:quote:approve", "quote:discount_policy:write"])
VIEWER = person("vic@co.example", ["sales"], [])
async def _api(method, path, headers, 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=headers, 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
if res.is_error:
return "refused", text.split(f"Error executing tool {name}: ", 1)[1]
return "ok", json.loads(text)
def tool(who, name, args):
return anyio.run(_tool, who, name, args)
def open_with(who, *lines):
status, q = api("POST", "/quotes", who, {"bp_id": ACME, "currency": "EUR",
"valid_until": "2099-12-31"})
assert status == 200, q
for product, qty, price in lines:
body = {"product_id": product, "quantity": qty}
if price is not None:
body["unit_price"] = price
status, q = api("POST", f"/quotes/{q['quote_id']}/versions/1/lines", who, body)
assert status == 200, q
return q
# ── CRM-05, end to end: priced from the price book, totals computed ────────────────────────
def test_crm05_a_quote_priced_from_the_price_book():
q = open_with(ANNA, ("AD-PAGE", 2, None), ("AD-HALF", 1, "570"), ("DIGITAL", 3, None))
assert q["total"] == "3320.00" and q["list_total"] == "3350.00"
assert q["discount_pct"] == "5.00" and q["net_discount_pct"] == "0.90" and q["status"] == "DRAFT"
assert q["lines"][1]["price_book_id"] == "standard"
def test_an_agent_builds_the_same_quote_through_mcp():
verdict, q = tool(ANNA, "open_quote", {"bp_id": ACME, "currency": "EUR"})
assert verdict == "ok"
verdict, q = tool(ANNA, "add_line", {"quote_id": q["quote_id"], "version": 1,
"product_id": "AD-PAGE", "quantity": 2})
assert verdict == "ok" and q["total"] == "2000.00"
def test_a_product_with_no_price_says_so():
q = open_with(ANNA)
status, body = api("POST", f"/quotes/{q['quote_id']}/versions/1/lines", ANNA,
{"product_id": "NOPE"})
assert status == 422 and "has no EUR price for NOPE" in body["detail"]
# ── CRM-06: the approval route, through both doors ─────────────────────────────────────────
def test_crm06_eighteen_percent_waits_for_the_desk_and_the_rep_cannot_approve():
q = open_with(ANNA, ("AD-PAGE", 1, "820"))
qid = q["quote_id"]
assert q["needs_approval"] is True
status, body = api("POST", f"/quotes/{qid}/versions/1/send", ANNA)
assert status == 403 and "18.00%" in body["detail"]
# the rep lacks the permission; the desk sees it in the queue
assert api("POST", f"/quotes/{qid}/versions/1/approve", ANNA)[0] == 403
assert [w["quote_id"] for w in api("GET", "/approvals", DESK)[1]["waiting"]] == [qid]
# the desk approves — through MCP, as an agent would — and the rep sends
verdict, approved = tool(DESK, "approve_discount", {"quote_id": qid, "version": 1})
assert verdict == "ok" and approved["approval"]["approved_pct"] == "18.00"
assert api("POST", f"/quotes/{qid}/versions/1/send", ANNA)[1]["status"] == "SENT"
assert api("GET", "/approvals", DESK)[1]["waiting"] == []
def test_crm06_above_the_maximum_is_refused_through_both_doors():
q = open_with(ANNA)
path = f"/quotes/{q['quote_id']}/versions/1/lines"
status, body = api("POST", path, ANNA, {"product_id": "AD-PAGE", "unit_price": "500"})
verdict, said = tool(ANNA, "add_line", {"quote_id": q["quote_id"], "version": 1,
"product_id": "AD-PAGE", "unit_price": "500"})
assert status == 409 and verdict == "refused"
assert said == f"409: {body['detail']}" and "maximum" in said
def test_the_policy_is_a_row_the_desk_changes():
assert api("PUT", "/discount-policy", DESK, {"auto_approve_pct": "20", "max_pct": "35"})[0] == 200
q = open_with(ANNA, ("AD-PAGE", 1, "820")) # 18%: now inside
assert q["needs_approval"] is False
assert api("PUT", "/discount-policy", ANNA, {"auto_approve_pct": "50", "max_pct": "60"})[0] == 403
# ── CRM-07: locked once sent; a change is a revision ───────────────────────────────────────
def test_crm07_a_sent_quote_refuses_edits_and_revises_into_v2():
q = open_with(ANNA, ("AD-PAGE", 1, None))
qid = q["quote_id"]
api("POST", f"/quotes/{qid}/versions/1/send", ANNA)
status, body = api("PATCH", f"/quotes/{qid}/versions/1/lines/1", ANNA, {"quantity": 2})
assert status == 409 and "Revise" in body["detail"]
status, v2 = api("POST", f"/quotes/{qid}/revise", ANNA, {})
assert status == 200 and (v2["version"], v2["status"], len(v2["lines"])) == (2, "DRAFT", 1)
v1 = api("GET", f"/quotes/{qid}", ANNA, params={"version": 1})[1]
assert v1["status"] == "SUPERSEDED"
def test_accepting_twice_is_one_acceptance():
q = open_with(ANNA, ("AD-PAGE", 1, None))
qid = q["quote_id"]
api("POST", f"/quotes/{qid}/versions/1/send", ANNA)
for _ in range(2):
status, body = api("POST", f"/quotes/{qid}/versions/1/decision", ANNA, {"outcome": "ACCEPTED"})
assert status == 200 and body["status"] == "ACCEPTED"
actions = [e["action"] for e in api("GET", f"/quotes/{qid}/history", ANNA)[1]["events"]]
assert actions.count("accepted") == 1
# ── who sees what ──────────────────────────────────────────────────────────────────────────
def test_another_reps_quote_does_not_exist_for_you():
q = open_with(ANNA, ("AD-PAGE", 1, None))
assert api("GET", f"/quotes/{q['quote_id']}", BEN)[0] == 404
assert tool(BEN, "get_quote", {"quote_id": q["quote_id"]}) == \
("refused", f"404: no quote {q['quote_id']}")
assert api("GET", "/quotes", BEN)[1] == {"quotes": []}
def test_the_desk_sees_every_quote_but_cannot_edit_one():
q = open_with(ANNA, ("AD-PAGE", 1, None))
assert api("GET", f"/quotes/{q['quote_id']}", DESK)[0] == 200
status, _ = api("PATCH", f"/quotes/{q['quote_id']}/versions/1/lines/1", DESK, {"quantity": 3})
assert status == 403 # lacks quote:quote:draft
def test_without_the_permission_nothing_is_written():
assert api("POST", "/quotes", VIEWER, {"bp_id": ACME, "currency": "EUR"})[0] == 403
assert api("GET", "/quotes", VIEWER)[1] == {"quotes": []}
# ── the record ─────────────────────────────────────────────────────────────────────────────
def test_the_history_names_who_did_what():
q = open_with(ANNA, ("AD-PAGE", 1, "820"))
tool(DESK, "approve_discount", {"quote_id": q["quote_id"], "version": 1})
events = api("GET", f"/quotes/{q['quote_id']}/history", ANNA)[1]["events"]
assert [(e["action"], e["actor"]) for e in events] == [
("opened", "anna@co.example"), ("line added", "anna@co.example"),
("discount approved", "dee@co.example")]
def test_a_price_is_never_changed_in_place():
status, body = api("PUT", "/prices/standard/AD-PAGE", DESK,
{"currency": "EUR", "valid_from": "2026-01-01", "list_price": "1100"})
assert status == 403 # the desk has no price-book permission …
admin = person("pat@co.example", ["pricing"], ["quote:price_book:write"])
status, body = api("PUT", "/prices/standard/AD-PAGE", admin,
{"currency": "EUR", "valid_from": "2026-01-01", "list_price": "1100"})
assert status == 409 and "later date" in body["detail"]
assert api("PUT", "/prices/standard/AD-PAGE", admin,
{"currency": "EUR", "valid_from": "2026-07-01", "list_price": "1100"})[0] == 200
found = api("GET", "/prices/standard/AD-PAGE", ANNA, params={"currency": "eur", "on": "2026-03-01"})
assert found[1]["list_price"] == "1000.00" # an earlier date keeps its price
def test_every_use_case_is_a_tool_with_honest_hints():
async def listed():
async with app.router.lifespan_context(app):
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=ANNA,
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}
tools = anyio.run(listed)
assert set(tools) == {u.name for u in use_cases.QUOTE.use_cases if u.tool}
assert tools["remove_line"].annotations.destructive_hint is True
assert tools["get_quote"].annotations.read_only_hint is True
assert tools["record_decision"].annotations.idempotent_hint is True
def test_seeing_a_quote_is_not_owning_it():
"""An analyst who also drafts sees every quote, but changes only their own; a manager may change
any. The permission alone would let both through — it is the ownership rule that decides."""
q = open_with(ANNA, ("AD-PAGE", 1, None))
path = f"/quotes/{q['quote_id']}/versions/1/lines/1"
analyst = person("ali@co.example", ["analyst"], REP + ["quote:quote:read_all"])
manager = person("max@co.example", ["manager"], REP + ["quote:quote:read_all", "quote:quote:manage"])
assert api("GET", f"/quotes/{q['quote_id']}", analyst)[0] == 200
status, body = api("PATCH", path, analyst, {"quantity": 3})
assert status == 403 and "only they or a manager" in body["detail"]
assert api("PATCH", path, manager, {"quantity": 3})[1]["lines"][0]["quantity"] == 3
# ── the customer, checked with the business-partner brick (tier 1: in this process) ────────
@pytest.fixture
def partners(monkeypatch):
"""The REAL partner brick, registered in-process, on its own SQLite store."""
brick_bp = 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
engine = create_engine("sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False})
bp_md.create_all(engine)
@contextmanager
def repo():
with engine.begin() as conn:
yield PartnerRepository(conn)
monkeypatch.setattr(bp_uc, "repository", repo)
steward = {"sub": "s", "email": "s@co.example", "roles": [], "permissions": ["bp:partner:write"],
"auth_source": "pool"}
acme = bp_uc.upsert_partner(steward, "acme.com", "Acme AB", "SE", roles=["customer"])
register_local("business-partner", bp_app)
yield {"acme": acme["bp_id"], "block": lambda: bp_uc.set_partner_status(
steward, acme["bp_id"], "blocked", "unpaid")}
unregister_local("business-partner")
def test_a_quote_for_a_known_customer_opens(partners):
status, q = api("POST", "/quotes", ANNA, {"bp_id": partners["acme"], "currency": "EUR"})
assert status == 200 and q["bp_id"] == str(partners["acme"])
events = api("GET", f"/quotes/{q['quote_id']}/history", ANNA)[1]["events"]
assert "not checked" not in events[0]["detail"]
def test_a_quote_for_an_unknown_customer_is_refused(partners):
status, body = api("POST", "/quotes", ANNA, {"bp_id": 999, "currency": "EUR"})
assert status == 422 and "no business partner 999" in body["detail"]
def test_a_blocked_customer_gets_no_new_quote_and_no_send(partners):
status, q = api("POST", "/quotes", ANNA, {"bp_id": partners["acme"], "currency": "EUR"})
api("POST", f"/quotes/{q['quote_id']}/versions/1/lines", ANNA, {"product_id": "AD-PAGE"})
partners["block"]() # blocked between drafting and sending
status, body = api("POST", f"/quotes/{q['quote_id']}/versions/1/send", ANNA)
assert status == 409 and "Acme AB is blocked" in body["detail"]
assert api("POST", "/quotes", ANNA, {"bp_id": partners["acme"], "currency": "EUR"})[0] == 409
def test_a_partner_brick_that_is_down_does_not_stop_a_sale(monkeypatch):
"""Installed but unreachable: the quote opens, and its history says the customer was not checked."""
monkeypatch.setattr(use_cases.PARTNERS, "base", "http://127.0.0.1:9") # nothing listens on 9
status, q = api("POST", "/quotes", ANNA, {"bp_id": 777, "currency": "EUR"})
assert status == 200
detail = api("GET", f"/quotes/{q['quote_id']}/history", ANNA)[1]["events"][0]["detail"]
assert "customer not checked" in detail
def test_the_pipeline_asks_for_facts_and_gets_no_contents():
status, q = api("POST", "/quotes", ANNA, {"bp_id": ACME, "currency": "EUR", "opportunity_id": "O-9",
"valid_until": "2099-12-31"})
qid = q["quote_id"]
api("POST", f"/quotes/{qid}/versions/1/lines", ANNA, {"product_id": "AD-PAGE", "quantity": 2})
pipeline_service = person("pipeline@svc", ["service"], ["quote:opportunity:read"])
facts = api("GET", "/opportunities/O-9/quote-facts", pipeline_service)[1]
assert (facts["has_sent_quote"], facts["has_accepted_quote"]) == (False, False)
api("POST", f"/quotes/{qid}/versions/1/send", ANNA)
api("POST", f"/quotes/{qid}/versions/1/decision", ANNA, {"outcome": "ACCEPTED"})
facts = api("GET", "/opportunities/O-9/quote-facts", pipeline_service)[1]
assert facts == {"opportunity_id": "O-9", "has_sent_quote": True, "has_accepted_quote": True,
"accepted_totals": {"EUR": "2000.00"},
"accepted": [{"quote_id": qid, "bp_id": str(ACME), "currency": "EUR", "total": "2000.00",
"version": 1}]}
assert api("GET", "/opportunities/O-9/quote-facts", BEN)[0] == 403 # a rep cannot ask this
# ── 0.2.0: the CRM audit's wave 1 (2026-10-17) ─────────────────────────────────────────────
async def _raw(method, path, headers, body=None):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://b") as c:
return await c.request(method, path, headers=headers, json=body)
def raw(method, path, headers, body=None):
return anyio.run(_raw, method, path, headers, body)
def test_h2_one_marked_up_line_cannot_hide_a_forty_percent_discount():
q = open_with(ANNA, ("AD-PAGE", 1, "1800"))
status, body = api("POST", f"/quotes/{q['quote_id']}/versions/1/lines", ANNA,
{"product_id": "AD-PAGE", "unit_price": "600"})
assert status == 409 and "40.00%" in body["detail"] and "maximum" in body["detail"]
marked_up = open_with(BEN, ("AD-PAGE", 1, "1250"))
assert marked_up["discount_pct"] == "0.00" and marked_up["net_discount_pct"] == "0.00"
assert marked_up["lines"][0]["discount_pct"] == "0.00"
def test_c3_a_quote_carries_its_revision_and_a_stale_one_is_409():
q = open_with(ANNA, ("AD-PAGE", 1, None))
qid, rev = q["quote_id"], q["revision"]
assert rev == 2 # opened (1), a line added (2)
assert raw("GET", f"/quotes/{qid}", ANNA).headers["etag"] == '"2"'
ok = api("PATCH", f"/quotes/{qid}/versions/1/lines/1", ANNA, {"quantity": 2, "revision": 2})[1]
assert ok["revision"] == 3
assert api("POST", f"/quotes/{qid}/versions/1/lines", ANNA, {"product_id": "DIGITAL", "revision": 2}) == \
(409, {"detail": "changed since you opened it — reload it and make your change again"})
assert tool(ANNA, "send_quote", {"quote_id": qid, "version": 1, "revision": 2})[1].startswith("409")
r = raw("POST", f"/quotes/{qid}/versions/1/send", {**ANNA, "If-Match": '"3"'})
assert r.status_code == 200 and r.json()["status"] == "SENT"
def test_c3_two_lines_added_at_once_without_a_revision_are_both_kept(monkeypatch):
"""The audit's eight concurrent adds kept ONE line: every add loaded the empty draft, numbered its line 1,
and the last save won. The second add here read the draft before the first committed (DataK3's snapshot);
it is re-run on the latest, so both lines are kept, numbered 1 and 2."""
from brick_quote.infrastructure.tables import QuoteRepository
q = open_with(ANNA)
with use_cases.repository() as r:
snapshot = r.get(q["quote_id"])
api("POST", f"/quotes/{q['quote_id']}/versions/1/lines", ANNA, {"product_id": "AD-PAGE"})
real, once = QuoteRepository.get, {"n": 1}
def stale(self, quote_id):
if once["n"] and quote_id == q["quote_id"]:
once["n"] -= 1
import copy
return copy.deepcopy(snapshot)
return real(self, quote_id)
monkeypatch.setattr(QuoteRepository, "get", stale)
status, after = api("POST", f"/quotes/{q['quote_id']}/versions/1/lines", ANNA, {"product_id": "DIGITAL"})
assert status == 200
assert [(l["line_no"], l["product_id"]) for l in after["lines"]] == [(1, "AD-PAGE"), (2, "DIGITAL")]
def test_c4_a_bp_id_leaves_as_a_string_and_comes_in_as_either():
status, q = api("POST", "/quotes", ANNA, {"bp_id": "45665172796086509", "currency": "EUR"})
assert status == 200 and q["bp_id"] == "45665172796086509"
assert tool(ANNA, "open_quote", {"bp_id": 45665172796086509, "currency": "EUR"})[1]["bp_id"] == \
"45665172796086509"
assert api("GET", "/quotes", ANNA)[1]["quotes"][0]["bp_id"] == "45665172796086509"
def test_h12_a_bad_input_is_422_never_500():
q = open_with(ANNA)
path = f"/quotes/{q['quote_id']}/versions/1/lines"
for body in ({"product_id": "AD-PAGE", "quantity": 2 ** 31}, {"product_id": "AD-PAGE", "quantity": 0},
{"product_id": "AD-PAGE", "unit_price": "1e20"}, {"product_id": "AD-PAGE", "description": "nul\x00"},
{"product_id": "AD-PAGE", "priced_on": "2026-02-30"}, {"product_id": "P" * 300}):
status, out = api("POST", path, ANNA, body)
assert status == 422, (body, status, out)
assert tool(ANNA, "add_line", {"quote_id": q["quote_id"], "version": 1, **body})[0] == "refused"
assert api("POST", "/quotes", ANNA, {"bp_id": ACME, "currency": "EUR", "valid_until": "yesterday"})[0] == 422
assert api("POST", "/quotes", ANNA, {"bp_id": "99999999999999999999999", "currency": "EUR"})[0] == 422
def test_h4_a_second_quote_for_a_deal_is_the_one_in_flight_said_so():
status, first = api("POST", "/quotes", ANNA, {"bp_id": ACME, "currency": "EUR", "opportunity_id": "O-1"})
assert status == 200 and first["existing"] is False
status, again = api("POST", "/quotes", ANNA, {"bp_id": ACME, "currency": "EUR", "opportunity_id": "O-1"})
assert status == 200 and again["existing"] is True and again["quote_id"] == first["quote_id"]
assert tool(ANNA, "open_quote", {"bp_id": ACME, "currency": "EUR", "opportunity_id": "O-1"})[1][
"quote_id"] == first["quote_id"]
# one the caller may not see is not silently doubled: 409, and no second quote
manager = person("max@co.example", ["manager"], REP + ["quote:quote:read_all", "quote:quote:manage"])
mine = api("POST", "/quotes", manager, {"bp_id": ACME, "currency": "EUR", "opportunity_id": "O-2"})[1]
status, body = api("POST", "/quotes", BEN, {"bp_id": ACME, "currency": "EUR", "opportunity_id": "O-2"})
assert status == 409 and "already has a quote in flight" in body["detail"]
assert [x["quote_id"] for x in api("GET", "/quotes", manager)[1]["quotes"] if x["quote_id"] == mine["quote_id"]]
assert api("GET", "/quotes", BEN)[1]["quotes"] == []
def test_h4_two_opens_for_one_deal_at_once_make_one_quote(monkeypatch):
"""The second open looked for a quote on the deal before the first committed (it saw none): its claim of
the deal's next quote collides, and it answers the winner's quote."""
from brick_quote.infrastructure.tables import QuoteRepository
first = api("POST", "/quotes", ANNA, {"bp_id": ACME, "currency": "EUR", "opportunity_id": "O-3"})[1]
real, once = QuoteRepository.ids, {"n": 1}
def blind(self, owner=None, opportunity_id=None):
if once["n"] and opportunity_id == "O-3":
once["n"] -= 1
return []
return real(self, owner=owner, opportunity_id=opportunity_id)
monkeypatch.setattr(QuoteRepository, "ids", blind)
status, second = api("POST", "/quotes", ANNA, {"bp_id": ACME, "currency": "EUR", "opportunity_id": "O-3"})
assert status == 200 and second["quote_id"] == first["quote_id"] and second["existing"] is True
assert len(api("GET", "/quotes", ANNA)[1]["quotes"]) == 1
def test_h13_a_retried_add_line_with_one_idempotency_key_adds_one_line(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.QUOTE, "idempotency_connection", e.begin)
q = open_with(ANNA)
path = f"/quotes/{q['quote_id']}/versions/1/lines"
a = raw("POST", path, {**ANNA, "Idempotency-Key": "line-1"}, {"product_id": "AD-PAGE"})
b = raw("POST", path, {**ANNA, "Idempotency-Key": "line-1"}, {"product_id": "AD-PAGE"})
assert a.json() == b.json() and b.headers["idempotent-replayed"] == "true"
assert len(api("GET", f"/quotes/{q['quote_id']}", ANNA)[1]["lines"]) == 1
# ── tier 1: the deal, asked of the REAL pipeline brick as the caller (C1) ──────────────────
@pytest.fixture
def crm(monkeypatch):
pytest.importorskip("brick_pipeline")
from erp_platform.seam import register_local, unregister_local
from brick_pipeline.app import app as pipe_app
from brick_pipeline.application import use_cases as p_uc
from brick_pipeline.infrastructure.tables import PipelineRepository, md as p_md
engine = create_engine("sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False})
p_md.create_all(engine)
@contextmanager
def repo():
with engine.begin() as conn:
yield PipelineRepository(conn)
monkeypatch.setattr(p_uc, "repository", repo)
register_local("pipeline", pipe_app)
rep = lambda email, roles=("sales",): {"sub": email, "email": email, "roles": list(roles), "auth_source": "pool",
"permissions": ["pipeline:opportunity:write"]}
yield {"p": p_uc, "rep": rep}
unregister_local("pipeline")
def test_c1_a_quote_is_attached_only_to_a_deal_its_rep_may_see_for_its_customer(crm):
anna_deal = crm["p"].open_opportunity(crm["rep"]("anna@co.example"), ACME, "Acme renewal", "EUR")
ben_deal = crm["p"].open_opportunity(crm["rep"]("ben@co.example"), ACME, "Ben's deal", "EUR")
other = crm["p"].open_opportunity(crm["rep"]("anna@co.example"), 777, "Other Co", "EUR")
sek = crm["p"].open_opportunity(crm["rep"]("anna@co.example"), ACME, "In crowns", "SEK")
body = lambda oid, cur="EUR": {"bp_id": ACME, "currency": cur, "opportunity_id": oid}
assert api("POST", "/quotes", ANNA, body(ben_deal["opportunity_id"])) == \
(404, {"detail": f"no opportunity {ben_deal['opportunity_id']}"})
assert api("POST", "/quotes", ANNA, body("O-NOPE"))[0] == 404
status, out = api("POST", "/quotes", ANNA, body(other["opportunity_id"]))
assert status == 409 and "partner" in out["detail"]
status, out = api("POST", "/quotes", ANNA, body(sek["opportunity_id"]))
assert status == 409 and "SEK" in out["detail"]
assert tool(ANNA, "open_quote", body(ben_deal["opportunity_id"]))[1].startswith("404")
assert api("GET", "/quotes", ANNA)[1]["quotes"] == [] # nothing was written
status, q = api("POST", "/quotes", ANNA, body(anna_deal["opportunity_id"]))
assert status == 200 and q["opportunity_id"] == anna_deal["opportunity_id"]
manager = person("max@co.example", ["manager"], REP + ["quote:quote:read_all", "quote:quote:manage",
"pipeline:opportunity:read_all"]) # the deal, asked as them
assert api("POST", "/quotes", manager, body(ben_deal["opportunity_id"]))[0] == 200 # a manager sees it
def test_c1_no_quote_for_a_closed_deal(crm):
d = crm["p"].open_opportunity(crm["rep"]("anna@co.example"), ACME, "Lost one", "EUR")
crm["p"].move_opportunity(crm["rep"]("anna@co.example"), d["opportunity_id"], "closed_lost", "price")
status, out = api("POST", "/quotes", ANNA, {"bp_id": ACME, "currency": "EUR", "opportunity_id": d["opportunity_id"]})
assert status == 409 and "closed" in out["detail"]
# ── 0.3.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_quote/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_quote.application.use_cases import QUOTE, READ_ALL, MANAGE
spec = yaml.safe_load((pathlib.Path(use_cases.__file__).parents[1] / "roles.yaml").read_text())
known = {uc.permission for uc in QUOTE.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(QUOTE.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 quote:quote:read_all sees every quote, and only quote:quote:manage changes another's."""
q = open_with(ANNA, ("AD-PAGE", 1, None))
path = f"/quotes/{q['quote_id']}/versions/1/lines/1"
named = person("nam@co.example", ["manager", "analyst"], REP)
assert api("GET", "/quotes", named)[1]["quotes"] == []
assert api("GET", f"/quotes/{q['quote_id']}", named)[0] == 404
reader = person("ops@co.example", ["revops"], REP + ["quote:quote:read_all"])
assert [x["quote_id"] for x in api("GET", "/quotes", reader)[1]["quotes"]] == [q["quote_id"]]
assert api("PATCH", path, reader, {"quantity": 2})[0] == 403
changer = person("lead@co.example", ["team_lead"], REP + ["quote:quote:read_all", "quote:quote:manage"])
assert api("PATCH", path, changer, {"quantity": 2})[1]["lines"][0]["quantity"] == 2
def test_the_process_is_served_as_bpmn():
status, view = api("GET", "/process/bpmn", ANNA)
assert status == 200, view
text = json.dumps(view)
for key in ("open_quote", "approve_discount", "send_quote", "accept_quote", "revise_quote"):
assert f"Task_{key}" in text, key