"""Tier 1: the lead brick with the REAL business-partner, activity and pipeline bricks running in this process
(erp_platform.seam.register_local), each on its own SQLite. Converting a lead (CRM-03) and excluding
customers from lookalikes (CRM-14) are tested across actual bricks, through their routes and their guards —
not stand-ins — including a neighbour that is down and a reply that is lost."""
import json
from decimal import Decimal
import anyio
import httpx
import pytest
from brick_lead.app import app
from brick_lead.application import use_cases as uc
from test_doors import BOSS, TOM, SOFIA, sqlite_repo
pytest.importorskip("brick_bp")
pytest.importorskip("brick_activity")
pytest.importorskip("brick_pipeline")
STEWARD = {"sub": "s", "email": "steward@co.example", "roles": [], "auth_source": "pool",
"permissions": ["bp:partner:write", "bp:family:write"]}
MANAGER = {"sub": "max@co.example", "email": "max@co.example", "roles": ["manager"], "auth_source": "pool",
"permissions": ["lead:lead:read_all", "lead:lead:manage", "lead:lead:assign",
"pipeline:opportunity:read_all", "pipeline:opportunity:manage"]}
@pytest.fixture
def suite(monkeypatch):
from erp_platform.seam import register_local, unregister_local
from brick_activity.app import app as act_app
from brick_activity.application import use_cases as act_uc
from brick_activity.infrastructure.tables import ActivityRepository, md as act_md
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_lead.infrastructure.tables import LeadRepository, md as lead_md
from brick_pipeline.app import app as pipe_app
from brick_pipeline.application import use_cases as pipe_uc
from brick_pipeline.infrastructure.tables import PipelineRepository, md as pipe_md
monkeypatch.setattr(uc, "repository", sqlite_repo(lead_md, LeadRepository))
monkeypatch.setattr(bp_uc, "repository", sqlite_repo(bp_md, PartnerRepository))
monkeypatch.setattr(act_uc, "repository", sqlite_repo(act_md, ActivityRepository))
monkeypatch.setattr(pipe_uc, "repository", sqlite_repo(pipe_md, PipelineRepository))
apps = {"business-partner": bp_app, "activity": act_app, "pipeline": pipe_app}
for name, brick in apps.items():
register_local(name, brick)
yield {"bp": bp_uc, "activity": act_uc, "pipeline": pipe_uc, "apps": apps, "register": register_local}
for name in apps:
unregister_local(name)
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()
def qualified(who=TOM, email="ana@acme.com", company="Acme AB", country="SE", **kw):
"""A lead worked and qualified the way a rep does it: engagement recorded, then qualify (score ≥ 60)."""
status, lead = api("POST", "/leads", who, {"email": email, "name": kw.pop("name", "Ana Berg"),
"company": company, "country": country, **kw})
assert status == 200, lead
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
svc = {"sub": "c@svc", "email": "c@svc", "roles": ["service"], "permissions": ["lead:signal:record"],
"auth_source": "pool"}
for kind in ("email_opened_pricing", "demo_requested"):
uc.record_signal(svc, lead["lead_id"], kind, now)
api("POST", f"/leads/{lead['lead_id']}/working", who)
status, lead = api("POST", f"/leads/{lead['lead_id']}/qualify", who)
assert status == 200 and lead["status"] == "qualified", lead
return lead
def convert(lead, who=TOM, **body):
return api("POST", f"/leads/{lead['lead_id']}/convert", who, {"currency": "EUR", **body})
def world(s, domain="acme.com", email="ana@acme.com"):
"""What the neighbours hold — read through their own use cases, as their owners see it."""
partners = s["bp"].find_partners(STEWARD, bp_key=domain)["partners"]
with s["activity"].repository() as r:
contact = r.contact_by_email(email)
deals = s["pipeline"].list_opportunities(MANAGER, open_only=False)["opportunities"]
return partners, contact, deals
def down(detail="restarting"):
"""A neighbour that is installed but not answering properly: every call is a 503."""
async def app_(scope, receive, send):
if scope["type"] != "http":
return
await send({"type": "http.response.start", "status": 503,
"headers": [(b"content-type", b"application/json")]})
await send({"type": "http.response.body", "body": json.dumps({"detail": detail}).encode()})
return app_
def reply_lost(real):
"""The neighbour does the work — and the reply never reaches the caller (a timeout, a dropped
connection): the caller sees a 502 for a deal that WAS opened. The retry is what must not duplicate it."""
async def app_(scope, receive, send):
if scope["type"] != "http":
return await real(scope, receive, send)
async def swallow(message):
pass
await real(scope, receive, swallow)
await send({"type": "http.response.start", "status": 502, "headers": [(b"content-type", b"text/plain")]})
await send({"type": "http.response.body", "body": b"bad gateway"})
return app_
# ── CRM-03 ─────────────────────────────────────────────────────────────────────────────────
def test_crm03_one_partner_one_contact_one_deal_in_the_first_stage_and_converting_again_changes_nothing(suite):
lead = qualified()
status, done = convert(lead, opportunity_name="Acme — first order", amount="12000")
assert status == 200 and done["status"] == "converted", done
partners, contact, deals = world(suite)
assert [(p["bp_id"], p["roles"], p["legal_name"], p["country"]) for p in partners] == \
[(int(done["bp_id"]), ["customer"], "Acme AB", "SE")]
assert (contact.contact_id, contact.bp_id, contact.full_name) == (done["contact_id"], int(done["bp_id"]), "Ana Berg")
assert [(d["opportunity_id"], d["bp_id"], d["stage"], d["owner"], d["amount"]) for d in deals] == \
[(done["opportunity_id"], int(done["bp_id"]), "discovery", "tom@co.example", Decimal("12000.00"))]
status, again = convert(lead)
assert status == 200 and {k: again[k] for k in ("bp_id", "contact_id", "opportunity_id")} == \
{k: done[k] for k in ("bp_id", "contact_id", "opportunity_id")} and "nothing changed" in again["note"]
assert [len(x) if isinstance(x, list) else 1 for x in world(suite)] == [1, 1, 1]
assert api("GET", f"/leads/{lead['lead_id']}", TOM)[1]["conversion"]["bp_id"] == done["bp_id"]
def test_crm22_an_existing_partner_for_the_domain_is_reused_and_becomes_a_customer(suite):
existing = suite["bp"].upsert_partner(STEWARD, "acme.com", "Acme Aktiebolag", "SE", roles=["supplier"])
status, done = convert(qualified())
assert status == 200 and int(done["bp_id"]) == existing["bp_id"]
partner = suite["bp"].get_partner(STEWARD, existing["bp_id"])
assert partner["roles"] == ["customer", "supplier"] and partner["legal_name"] == "Acme Aktiebolag"
def test_crm03_pipeline_down_after_partner_and_contact_keeps_them_and_a_retry_finishes_without_duplicates(suite):
lead = qualified()
suite["register"]("pipeline", down("deploying"))
status, body = convert(lead)
assert status == 503 and "the pipeline did not answer (deploying)" in body["detail"]
assert "Done so far: partner" in body["detail"] and "contact C-" in body["detail"]
kept = api("GET", f"/leads/{lead['lead_id']}", TOM)[1]
assert kept["status"] == "qualified" and kept["conversion"]["bp_id"] and kept["conversion"]["contact_id"]
assert kept["conversion"]["opportunity_id"] is None
suite["register"]("pipeline", suite["apps"]["pipeline"])
status, done = convert(lead)
assert status == 200 and done["status"] == "converted"
assert (done["bp_id"], done["contact_id"]) == (kept["conversion"]["bp_id"], kept["conversion"]["contact_id"])
partners, contact, deals = world(suite)
assert (len(partners), contact.contact_id, len(deals)) == (1, done["contact_id"], 1)
steps = [e["action"] for e in api("GET", f"/leads/{lead['lead_id']}/history", TOM)[1]["events"]]
assert steps[-3:] == ["conversion step recorded", "conversion step recorded", "converted"]
def test_crm03_a_deal_whose_reply_was_lost_is_not_opened_twice_on_the_retry(suite):
"""The idempotency gap closed in brick-pipeline 0.2.0 (source_ref): without it this makes two deals."""
lead = qualified()
suite["register"]("pipeline", reply_lost(suite["apps"]["pipeline"]))
status, body = convert(lead)
assert status == 503 and "the pipeline did not answer (bad gateway)" in body["detail"]
assert len(world(suite)[2]) == 1 # the deal exists; the lead does not know it yet
suite["register"]("pipeline", suite["apps"]["pipeline"])
status, done = convert(lead)
assert status == 200 and done["status"] == "converted"
deals = world(suite)[2]
assert [d["opportunity_id"] for d in deals] == [done["opportunity_id"]]
def test_crm03_a_partner_register_that_refuses_creates_nothing_and_says_why(suite):
lead = qualified(country="") # a new partner needs a country
status, body = convert(lead)
assert status == 422 and "the partner register refused" in body["detail"] and "country" in body["detail"]
assert "Nothing was created" in body["detail"]
assert world(suite)[0] == [] and world(suite)[2] == []
api("PATCH", f"/leads/{lead['lead_id']}", TOM, {"country": "SE"})
assert convert(lead)[1]["status"] == "converted"
def test_crm03_another_rep_cannot_convert_my_lead_and_a_blocked_customer_gets_no_deal(suite):
lead = qualified()
assert convert(lead, who=SOFIA)[0] == 404
bp_id = suite["bp"].upsert_partner(STEWARD, "acme.com", "Acme AB", "SE", roles=["customer"])["bp_id"]
suite["bp"].set_partner_status(STEWARD, bp_id, "blocked", "unpaid invoices")
status, body = convert(lead)
assert status == 409 and "the pipeline refused" in body["detail"] and "blocked" in body["detail"]
assert "Done so far: partner" in body["detail"] and world(suite)[2] == []
# ── CRM-14 across bricks ───────────────────────────────────────────────────────────────────
def test_crm14_lookalikes_exclude_customers_the_partner_register_holds_and_those_converted_from_leads(suite):
won = qualified(email="ana@acme.com", industry="software", size_band="51-200")
done = convert(won)[1]
for email, company in (("a@twin.se", "Twin"), ("b@held.se", "Held"), ("c@other.se", "Other")):
api("POST", "/leads", TOM, {"email": email, "company": company, "industry": "software",
"size_band": "51-200", "country": "SE"})
api("POST", "/leads", TOM, {"email": "d@acme.com", "company": "Acme AB", "industry": "software",
"size_band": "51-200", "country": "SE"}) # a second person at the customer
suite["bp"].upsert_partner(STEWARD, "held.se", "Held AB", "SE", roles=["customer"]) # a customer, no lead
suite["bp"].upsert_partner(STEWARD, "other.se", "Other AB", "SE", roles=["supplier"]) # NOT a customer
status, r = api("GET", "/lookalikes", TOM, params={"bp_id": done["bp_id"]})
assert status == 200 and r["note"] is None
assert [c["domain"] for c in r["lookalikes"]] == ["other.se", "twin.se"]
assert r["excluded_customers"] == 2 # acme.com (converted) and held.se (register)
suite["register"]("business-partner", down())
r = api("GET", "/lookalikes", TOM, params={"bp_id": done["bp_id"]})[1]
assert [c["domain"] for c in r["lookalikes"]] == ["held.se", "other.se", "twin.se"]
assert "did not answer" in r["note"]
assert api("GET", "/lookalikes", BOSS, params={"bp_id": done["bp_id"]})[0] == 200
# ── 0.2.0: the CRM audit's wave 1 (2026-10-17) ─────────────────────────────────────────────
def person_headers(email):
import base64
payload = base64.urlsafe_b64encode(json.dumps({"permissions": ["pipeline:opportunity:write"]}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": ["sales"]}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
LEAD_SERVICE = {"sub": "lead@svc", "email": "lead@svc", "roles": ["service"], "auth_source": "pool",
"permissions": ["pipeline:opportunity:link_source"]}
def test_c2_a_rep_cannot_plant_the_deal_another_reps_conversion_lands_on(suite):
"""The audit: Tom opened a deal under the source_ref of Sofia's lead, and her conversion landed on it."""
lead = qualified(SOFIA, email="mia@nordco.example", company="Nordco ApS", country="DK")
elsewhere = suite["bp"].upsert_partner(STEWARD, "elsewhere.example", "Elsewhere AB", "SE", roles=["customer"])
async def plant():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=suite["apps"]["pipeline"]),
base_url="http://p") as c:
return await c.post("/opportunities", headers=person_headers("tom@co.example"),
json={"bp_id": elsewhere["bp_id"], "name": "planted by tom", "currency": "EUR",
"source_ref": f"lead:{lead['lead_id']}"})
planted = anyio.run(plant).json()
assert planted["opportunity_id"] # a deal — but the ref was not taken
status, done = convert(lead, who=SOFIA, amount="30000")
assert status == 200 and done["opportunity_id"] != planted["opportunity_id"]
deal = suite["pipeline"].get_opportunity(MANAGER, done["opportunity_id"])
assert (deal["owner"], deal["bp_id"], deal["amount"]) == ("sofia@co.example", int(done["bp_id"]), Decimal("30000.00"))
def test_c2_a_ref_whose_deal_is_another_partners_stops_the_conversion_and_writes_nothing_after_it(suite):
lead = qualified()
other = suite["bp"].upsert_partner(STEWARD, "elsewhere.example", "Elsewhere AB", "SE", roles=["customer"])
poisoned = suite["pipeline"].open_opportunity_from_source(LEAD_SERVICE, f"lead:{lead['lead_id']}",
other["bp_id"], "not this lead's", "EUR", "x@co.example")
status, body = convert(lead)
assert status == 409 and "the pipeline refused" in body["detail"] and "another partner" not in body["detail"]
assert "for partner" in body["detail"] and "Done so far: partner" in body["detail"]
kept = api("GET", f"/leads/{lead['lead_id']}", TOM)[1]
assert kept["status"] == "qualified" and kept["conversion"]["opportunity_id"] is None
assert kept["conversion"]["bp_id"] and kept["conversion"]["contact_id"]
assert [d["opportunity_id"] for d in world(suite)[2]] == [poisoned["opportunity_id"]]
def test_a_half_converted_lead_resumes_once_the_cause_is_gone(suite):
"""Blocked when the deal was to be opened: the partner and contact are kept; unblocked, converting again
finishes with the SAME partner and contact."""
lead = qualified()
bp_id = suite["bp"].upsert_partner(STEWARD, "acme.com", "Acme AB", "SE", roles=["customer"])["bp_id"]
suite["bp"].set_partner_status(STEWARD, bp_id, "blocked", "unpaid invoices")
assert convert(lead)[0] == 409
kept = api("GET", f"/leads/{lead['lead_id']}", TOM)[1]["conversion"]
suite["bp"].set_partner_status(STEWARD, bp_id, "active", "paid")
status, done = convert(lead)
assert status == 200 and done["status"] == "converted"
assert (done["bp_id"], done["contact_id"]) == (kept["bp_id"], kept["contact_id"])
partners, contact, deals = world(suite)
assert (len(partners), len(deals)) == (1, 1)