dodil/erp-brick-lead

tests/test_doors.py

"""The lead brick through both doors — HTTP and MCP — over the REAL repository on SQLite.

The neighbours (business-partner, activity, pipeline) are NOT installed here, which is itself a case: converting is refused
before anything is created. tests/test_suite.py runs the real neighbours in this process (tier 1)."""
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_lead.app import app
from brick_lead.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_lead.infrastructure.tables import LeadRepository, md
    monkeypatch.setattr(use_cases, "repository", sqlite_repo(md, LeadRepository))


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 = ["lead:lead:write"]
TOM = person("tom@co.example", ["sales"], REP)
SOFIA = person("sofia@co.example", ["sales"], REP)
BOSS = person("max@co.example", ["manager"], REP + ["lead:lead:assign", "lead:policy:write", "lead:lead:read_all",
                                                   "lead:lead:manage"])          # 0.3.0: permissions, not the role name
MIA = person("mia@co.example", ["marketing"], ["lead:lead:import", "lead:lead:read_all"])
JOB = person("scoring@svc", ["service"], ["lead:lead:score"])
CAMPAIGNS = person("campaign@svc", ["service"], ["lead:signal:record"])


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 add(who=TOM, email="ana@acme.com", **kw):
    status, lead = api("POST", "/leads", who, {"email": email, "name": kw.pop("name", "Ana Berg"),
                                               "company": kw.pop("company", "Acme"), **kw})
    assert status == 200, lead
    return lead


def signal(lead, kind="email_opened_pricing", at="2026-10-17T11:00:00Z"):
    return api("POST", f"/leads/{lead['lead_id']}/signals", CAMPAIGNS, {"kind": kind, "at": at})


# ── CRM-13: import ─────────────────────────────────────────────────────────────────────────

TRADE_SHOW = """name,email,company,website,country,size
Ana Berg,ana@acme.com,Acme,acme.com,SE,51-200
Ben Ode,,Beta,,,
Cy Lund,cy@cyco.se,Cyco,,SE,250
"""


def test_crm13_importing_twice_each_person_exists_once_and_the_row_without_email_is_reported():
    status, first = api("POST", "/leads/import", MIA, {"csv_text": TRADE_SHOW, "source": "Nordic Expo"})
    assert status == 200, first
    assert (first["created"], first["updated"], first["unchanged"]) == (2, 0, 0)
    assert first["skipped"] == [{"row": 3, "reason": "no email"}]
    verdict, again = anyio.run(_tool, MIA, "import_leads", {"csv_text": TRADE_SHOW})
    assert verdict == "ok" and (again["created"], again["updated"], again["unchanged"]) == (0, 0, 2)
    assert again["skipped"] == first["skipped"] and again["lead_ids"] == first["lead_ids"]
    leads = api("GET", "/leads", MIA)[1]["leads"]
    assert sorted(l["email"] for l in leads) == ["ana@acme.com", "cy@cyco.se"]
    assert {l["source"] for l in leads} == {"nordic expo"} and {l["owner"] for l in leads} == {"mia@co.example"}


def test_a_reimport_updates_what_the_file_says_but_never_the_owner_or_status():
    api("POST", "/leads/import", MIA, {"csv_text": TRADE_SHOW})
    lead = api("GET", "/leads", MIA)[1]["leads"][0]
    api("PUT", f"/leads/{lead['lead_id']}/owner", BOSS, {"owner": "tom@co.example"})
    api("POST", f"/leads/{lead['lead_id']}/working", TOM)
    status, r = api("POST", "/leads/import", MIA, {"csv_text": f"email,industry\n{lead['email']},software\n"})
    assert (r["updated"], r["created"]) == (1, 0)
    after = api("GET", f"/leads/{lead['lead_id']}", TOM)[1]
    assert (after["industry"], after["owner"], after["status"], after["company"]) == \
        ("software", "tom@co.example", "working", lead["company"])


def test_only_an_importer_imports_and_a_file_without_emails_is_refused_whole():
    assert api("POST", "/leads/import", TOM, {"csv_text": TRADE_SHOW})[0] == 403
    status, body = api("POST", "/leads/import", MIA, {"csv_text": "name,company\nAna,Acme\n"})
    assert status == 422 and "email column" in body["detail"]


# ── CRM-01: who sees which lead ────────────────────────────────────────────────────────────

def test_crm01_a_rep_sees_only_their_own_leads_and_another_reps_is_not_found():
    mine, theirs = add(TOM), add(SOFIA, "bo@beta.io", company="Beta")
    assert [l["lead_id"] for l in api("GET", "/leads", TOM)[1]["leads"]] == [mine["lead_id"]]
    assert api("GET", f"/leads/{theirs['lead_id']}", TOM)[0] == 404
    assert api("POST", f"/leads/{theirs['lead_id']}/working", TOM)[0] == 404
    verdict, said = anyio.run(_tool, TOM, "get_lead", {"lead_id": theirs["lead_id"]})
    assert verdict == "refused" and said.startswith("404")
    assert len(api("GET", "/leads", BOSS)[1]["leads"]) == len(api("GET", "/leads", MIA)[1]["leads"]) == 2


def test_adding_someone_elses_lead_again_is_refused_not_duplicated():
    add(TOM)
    status, body = api("POST", "/leads", SOFIA, {"email": "ANA@acme.com"})
    assert status == 409 and "tom@" not in body["detail"]
    again = add(TOM, "ana@acme.com", name="other")
    assert again["existing"] and again["name"] == "Ana Berg"


def test_marketing_sees_every_lead_but_works_none_and_no_identity_is_401():
    lead = add(TOM)
    assert api("GET", f"/leads/{lead['lead_id']}", MIA)[0] == 200
    assert api("POST", f"/leads/{lead['lead_id']}/working", MIA)[0] == 403
    assert api("GET", "/leads", {})[0] == 401


# ── the workflow and CRM-10 ────────────────────────────────────────────────────────────────

def test_qualifying_is_gated_by_the_score_computed_now():
    lead = add(TOM)
    status, body = api("POST", f"/leads/{lead['lead_id']}/qualify", TOM)
    assert status == 409 and "start working it first" in body["detail"]
    api("POST", f"/leads/{lead['lead_id']}/working", TOM)
    status, body = api("POST", f"/leads/{lead['lead_id']}/qualify", TOM)
    assert status == 409 and "the score is 0, below the 60" in body["detail"]
    from datetime import datetime, timezone
    now = datetime.now(timezone.utc).isoformat()
    for kind in ("email_opened_pricing", "demo_requested"):
        assert signal(lead, kind, now)[1]["recorded"]
    verdict, q = anyio.run(_tool, TOM, "qualify_lead", {"lead_id": lead["lead_id"]})
    assert verdict == "ok" and q["status"] == "qualified"
    events = api("GET", f"/leads/{lead['lead_id']}/history", TOM)[1]["events"]
    assert "score 65 ≥ 60" in events[-1]["detail"]


def test_disqualifying_needs_a_reason_through_both_doors():
    lead = add(TOM)
    status, body = api("POST", f"/leads/{lead['lead_id']}/disqualify", TOM, {"reason": " "})
    assert status == 422
    verdict, said = anyio.run(_tool, TOM, "disqualify_lead", {"lead_id": lead["lead_id"], "reason": " "})
    assert said == f"422: {body['detail']}"
    assert api("POST", f"/leads/{lead['lead_id']}/disqualify", TOM, {"reason": "no budget"})[1]["status"] == \
        "disqualified"


def test_crm10_a_run_reflects_a_signal_is_recorded_and_running_it_twice_changes_nothing():
    lead = add(TOM, size_band="201-1000")
    api("POST", f"/leads/{lead['lead_id']}/working", TOM)
    cold = add(SOFIA, "bo@beta.io", company="Beta")
    first = signal(lead)[1]
    assert first["recorded"] and first["weighed_by_policy"]
    assert not signal(lead)[1]["recorded"]                             # the same event again: one signal
    assert signal(lead, "demo_requested", "2026-10-17T10:00:00+02:00")[0] == 200
    status, run = api("POST", "/leads/score", JOB, {"as_of": "2026-10-17T12:00:00Z"})
    assert status == 200 and (run["leads_scored"], run["scores_changed"]) == (2, 2)
    assert run["qualified"] == [lead["lead_id"]]                       # 15 + 25 + 40 = 80 ≥ 60, and working
    scored = api("GET", f"/leads/{lead['lead_id']}", TOM)[1]
    assert scored["score"] == 80 and scored["scored_by_run"] == run["run_id"] and scored["status"] == "qualified"
    assert scored["score_reasons"] == ["size_band=201-1000 +15", "email_opened_pricing ×1 +25",
                                       "demo_requested ×1 +40"]
    verdict, again = anyio.run(_tool, JOB, "score_leads", {"as_of": "2026-10-17T12:00:00Z"})
    assert verdict == "ok" and (again["scores_changed"], again["qualified"]) == (0, [])
    assert api("GET", f"/leads/{lead['lead_id']}", TOM)[1]["scored_by_run"] == run["run_id"]
    runs = api("GET", "/score-runs", BOSS)[1]["runs"]
    assert {r["run_id"] for r in runs} == {run["run_id"], again["run_id"]}
    assert api("GET", f"/leads/{cold['lead_id']}", SOFIA)[1]["score"] == 0
    assert api("POST", "/leads/score", TOM, {})[0] == 403
    assert api("GET", "/leads", TOM, params={"min_score": 50})[1]["leads"][0]["lead_id"] == lead["lead_id"]


def test_the_scoring_policy_is_the_customers_and_changes_the_next_run():
    lead = add(TOM, industry="software")
    policy = {"qualify_threshold": 30, "signal_window_days": 14,
              "fit": [{"attribute": "industry", "value": "SaaS", "points": 35}],
              "signals": [{"kind": "webinar_attended", "points": 20, "max_count": 1}]}
    assert api("PUT", "/scoring-policy", TOM, policy)[0] == 403
    status, saved = api("PUT", "/scoring-policy", BOSS, policy)
    assert status == 200 and saved["fit"] == [{"attribute": "industry", "value": "software", "points": 35}]
    assert api("GET", "/scoring-policy", TOM)[1]["qualify_threshold"] == 30
    api("POST", "/leads/score", JOB, {})
    assert api("GET", f"/leads/{lead['lead_id']}", TOM)[1]["score"] == 35
    bad = {**policy, "fit": [{"attribute": "shoe_size", "value": "9", "points": 5}]}
    assert api("PUT", "/scoring-policy", BOSS, bad)[0] == 422


# ── CRM-03 without neighbours ──────────────────────────────────────────────────────────────

def test_crm03_without_the_neighbours_converting_is_refused_before_anything_is_created():
    lead = add(TOM)
    assert api("POST", f"/leads/{lead['lead_id']}/convert", TOM, {"currency": "EUR"})[0] == 409   # not qualified
    with use_cases.repository() as repo:
        l = repo.get(lead["lead_id"])
        l.start_working()
        from brick_lead.domain.lead import Score, ScoringPolicy
        l.qualify(Score(99, 0, 99, ()), ScoringPolicy(60, 30, ()), "test")
        repo.save(l, "test")
    status, body = api("POST", f"/leads/{lead['lead_id']}/convert", TOM, {"currency": "EUR"})
    assert status == 409 and "business-partner, activity, pipeline" in body["detail"] and "Nothing was created" in body["detail"]
    assert api("GET", f"/leads/{lead['lead_id']}", TOM)[1]["conversion"]["bp_id"] is None


# ── CRM-14 on SQLite (ranked in Python; DataK3 ranks with <=>: scripts/live_check.py) ──────

def _lookalike_book():
    add(TOM, "a@nordsoft.se", company="Nordsoft", industry="software", size_band="201-1000", country="NO")
    add(TOM, "b@bigretail.com", company="Big Retail", industry="retail", size_band="5000+", country="US")
    add(TOM, "c@samesoft.se", company="Samesoft", industry="software", size_band="51-200", country="SE")
    add(TOM, "d@samesoft.se", name="Dee", company="Samesoft", industry="software", size_band="51-200", country="SE")
    add(SOFIA, "e@sofiasoft.se", company="Sofiasoft", industry="software", size_band="51-200", country="SE")
    won = add(TOM, "f@customer.se", company="Customer", industry="software", size_band="51-200", country="SE")
    with use_cases.repository() as repo:                     # as a finished conversion leaves it
        l = repo.get(won["lead_id"])
        l.status, l.bp_id, l.contact_id, l.opportunity_id = "converted", 54848654143250435, "C-1", "O-1"
        repo.save(l, "test")
    return won


def test_crm14_lookalikes_of_a_won_account_are_ranked_visible_and_exclude_customers():
    _lookalike_book()
    status, r = api("GET", "/lookalikes", TOM, params={"bp_id": 54848654143250435})
    assert status == 200 and r["based_on"]["industry"] == "software" and r["based_on"]["from"].startswith("lead ")
    assert [c["domain"] for c in r["lookalikes"]] == ["samesoft.se", "nordsoft.se", "bigretail.com"]
    assert r["lookalikes"][0]["similarity"] == 1.0 and len(r["lookalikes"][0]["leads"]) == 2
    assert "customer.se" not in [c["domain"] for c in r["lookalikes"]]         # the won account itself
    boss = api("GET", "/lookalikes", BOSS, params={"bp_id": 54848654143250435, "limit": 2})[1]
    assert [c["domain"] for c in boss["lookalikes"]] == ["samesoft.se", "sofiasoft.se"]   # a manager sees Sofia's
    assert "no partner register installed" in boss["note"]


def test_crm14_lookalikes_from_attributes_through_mcp_and_an_unknown_account_is_422():
    _lookalike_book()
    verdict, r = anyio.run(_tool, SOFIA, "find_lookalikes", {"industry": "software", "size_band": "51-200",
                                                             "country": "SE"})
    assert verdict == "ok" and [c["domain"] for c in r["lookalikes"]] == ["sofiasoft.se"]
    status, body = api("GET", "/lookalikes", TOM, params={"bp_id": 7})
    assert status == 422 and "pass industry" in body["detail"]



# ── 0.2.0: the CRM audit's wave 1 (2026-10-17) ─────────────────────────────────────────────

def tool(who, name, args):
    return anyio.run(_tool, who, name, args)


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 test_c3_a_lead_carries_its_version_and_a_stale_one_is_409():
    lead = add()
    assert lead["version"] == 1
    path = f"/leads/{lead['lead_id']}"
    assert api("PATCH", path, TOM, {"title": "CMO", "version": 1})[1]["version"] == 2
    assert api("PATCH", path, TOM, {"title": "CEO", "version": 1}) == \
        (409, {"detail": "changed since you opened it — reload it and make your change again"})
    assert tool(BOSS, "assign_lead", {"lead_id": lead["lead_id"], "owner": "sofia@co.example", "version": 1})[1] \
        .startswith("409: changed since you opened it")
    assert anyio.run(_raw, "GET", path, TOM).headers["etag"] == '"2"'
    r = anyio.run(_raw, "POST", f"{path}/working", {**TOM, "If-Match": '"2"'})
    assert r.status_code == 200 and r.json()["status"] == "working"


def test_c3_an_edit_and_an_assignment_at_once_without_a_version_both_land(monkeypatch):
    from brick_lead.infrastructure.tables import LeadRepository
    lead = add()
    with use_cases.repository() as r:
        snapshot = r.get(lead["lead_id"])
    api("PATCH", f"/leads/{lead['lead_id']}", TOM, {"title": "CMO"})
    real, once = LeadRepository.get, {"n": 1}

    def stale(self, lead_id):
        if once["n"] and lead_id == lead["lead_id"]:
            once["n"] -= 1
            import copy
            return copy.deepcopy(snapshot)
        return real(self, lead_id)
    monkeypatch.setattr(LeadRepository, "get", stale)
    assert api("PUT", f"/leads/{lead['lead_id']}/owner", BOSS, {"owner": "sofia@co.example"})[0] == 200
    now = api("GET", f"/leads/{lead['lead_id']}", BOSS)[1]
    assert (now["title"], now["owner"], now["version"]) == ("CMO", "sofia@co.example", 3)


def test_c3_a_reimport_that_raced_an_edit_loses_neither(monkeypatch):
    """An import read the lead before a rep's edit committed (DataK3's snapshot); its version claim collides,
    and the import is re-run on the latest: the file's change AND the rep's both stand."""
    from brick_lead.infrastructure.tables import LeadRepository
    add(email="ana@acme.com")
    with use_cases.repository() as r:
        snapshot = r.by_ids({use_cases.lead_id_for("ana@acme.com")})
    api("PATCH", f"/leads/{use_cases.lead_id_for('ana@acme.com')}", TOM, {"title": "CMO"})
    real, once = LeadRepository.by_ids, {"n": 1}

    def stale(self, ids):
        if once["n"]:
            once["n"] -= 1
            import copy
            return copy.deepcopy(snapshot)
        return real(self, ids)
    monkeypatch.setattr(LeadRepository, "by_ids", stale)
    status, out = api("POST", "/leads/import", MIA, {"csv_text": "email,country\nana@acme.com,NO\n"})
    assert status == 200 and out["updated"] == 1, out
    now = api("GET", f"/leads/{use_cases.lead_id_for('ana@acme.com')}", TOM)[1]
    assert (now["title"], now["country"]) == ("CMO", "NO")


def test_h12_a_bad_input_is_422_never_500():
    big = "name,email\n" + "x" * 200_000 + ",big@acme.com\n"
    status, out = api("POST", "/leads/import", MIA, {"csv_text": big})
    assert status == 422 and "row 2" in json.dumps(out), out
    assert tool(MIA, "import_leads", {"csv_text": big})[0] == "refused"
    assert api("POST", "/leads/import", MIA, {"csv_text": "name,email\nnul\x00,a@b.se\n"})[0] == 422
    assert api("POST", "/leads", TOM, {"email": "a@acme.com", "name": "N" * 300})[0] == 422
    lead = add()
    assert signal(lead, at="yesterday")[0] == 422
    assert api("POST", "/leads/score", JOB, {"as_of": "not-a-time"})[0] == 422
    assert api("GET", "/lookalikes", TOM, params={"bp_id": "99999999999999999999999"})[0] == 422
    assert api("GET", "/leads", TOM, params={"min_score": 1000})[0] == 422


def test_c4_ids_in_a_conversion_are_strings():
    lead = add()
    with use_cases.repository() as r:
        l = r.get(lead["lead_id"])
        l.bp_id = 45665172796086509
        r.save(l, "test")
    assert api("GET", f"/leads/{lead['lead_id']}", TOM)[1]["conversion"]["bp_id"] == "45665172796086509"


def test_a_half_converted_lead_keeps_the_keys_its_partner_and_contact_were_made_from():
    """A lead whose partner (from its domain) and contact (from its email) exist, and whose deal does not yet:
    converting again finishes it. Changing the domain or email in between would make the next step another
    company's or another person's."""
    lead = add()
    with use_cases.repository() as r:
        l = r.get(lead["lead_id"])
        l.status, l.domain, l.bp_id = "qualified", "acme.com", 4242
        r.save(l, "test")
    status, body = api("PATCH", f"/leads/{lead['lead_id']}", TOM, {"domain": "other.example"})
    assert status == 409 and "being converted" in body["detail"]
    assert api("PATCH", f"/leads/{lead['lead_id']}", TOM, {"title": "CMO"})[0] == 200        # the rest is editable


# ── 0.2.1: scoring and signals are MACHINE permissions — no wildcard reaches them (erp-platform 0.9.2) ──────

def test_a_wildcard_grant_never_reaches_scoring_or_signals():
    """`*:*:*`, `lead:*:*` and `lead:lead:*` run every person command and neither machine one: a score decides who
    gets qualified, and a signal moves it — only an exact grant, a service client's, passes."""
    lead = api("POST", "/leads", TOM, {"email": "wild@acme.example"})[1]
    for grant in ("*:*:*", "lead:*:*", "lead:lead:*", "lead:signal:*"):
        star = person("star@co.example", ["anything"], [grant])
        assert api("POST", "/leads/score", star, {}) == (403, {"detail": "missing permission: lead:lead:score"}), grant
        assert api("POST", f"/leads/{lead['lead_id']}/signals", star,
                   {"kind": "email_open", "at": "2026-10-17T09:00:00Z"}) == \
            (403, {"detail": "missing permission: lead:signal:record"}), grant
    assert api("POST", "/leads", person("star@co.example", ["anything"], ["*:*:*"]),
               {"email": "star-made@acme.example"})[0] == 200          # the person commands still pass
    assert api("POST", "/leads/score", JOB, {})[0] == 200                # the exact grants pass
    assert api("POST", f"/leads/{lead['lead_id']}/signals", CAMPAIGNS,
               {"kind": "email_open", "at": "2026-10-17T09:00:00Z"})[0] == 200


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 ("add_lead", "import_leads", "start_working", "qualify_lead", "disqualify_lead", "convert_lead"):
        assert f"Task_{key}" in text, key


# ── 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_lead/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_lead.application.use_cases import LEAD, READ_ALL, MANAGE
    spec = yaml.safe_load((pathlib.Path(use_cases.__file__).parents[1] / "roles.yaml").read_text())
    known = {uc.permission for uc in LEAD.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(LEAD.machine_permissions), role


def test_a_role_name_decides_nothing():
    """The old role names (`manager`, `analyst`, `marketing`) without the permissions see and change only their own;
    any role holding lead:lead:read_all sees every lead, only lead:lead:manage changes another's, and only
    lead:lead:assign names another owner for a new one."""
    toms = add()
    named = person("nam@co.example", ["manager", "analyst", "marketing"], REP)
    assert api("GET", "/leads", named)[1]["leads"] == []
    assert api("GET", f"/leads/{toms['lead_id']}", named)[0] == 404
    assert add(named, email="mine@acme.example", owner="tom@co.example")["owner"] == "nam@co.example"
    reader = person("ops@co.example", ["revops"], REP + ["lead:lead:read_all"])
    assert toms["lead_id"] in {x["lead_id"] for x in api("GET", "/leads", reader)[1]["leads"]}
    assert api("PATCH", f"/leads/{toms['lead_id']}", reader, {"name": "x"})[0] == 403
    changer = person("lead@co.example", ["team_lead"], REP + ["lead:lead:read_all", "lead:lead:manage", "lead:lead:assign"])
    assert api("PATCH", f"/leads/{toms['lead_id']}", changer, {"name": "x"})[0] == 200
    assert add(changer, email="theirs@acme.example", owner="sofia@co.example")["owner"] == "sofia@co.example"