dodil/erp-brick-activity

tests/test_doors.py

"""Contacts and activities through both doors over the REAL repository on SQLite — and, last, with the REAL
pipeline and business-partner bricks 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_activity.app import app
from brick_activity.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_activity.infrastructure.tables import ActivityRepository, md
    monkeypatch.setattr(use_cases, "repository", sqlite_repo(md, ActivityRepository))


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 = ["activity:contact:write", "activity:activity:write", "pipeline:opportunity:write"]
TOM, SOFIA = person("tom@co.example", ["sales"], REP), person("sofia@co.example", ["sales"], REP)
BOSS = person("max@co.example", ["manager"], REP + ["activity:contact:manage", "activity:activity:manage",
                                                   "pipeline:opportunity:read_all"])   # sees every deal, asked as them
READER = person("ann@co.example", ["sales"], [])
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 contact(who=TOM, email="anna@acme.com", bp_id=ACME, name="Anna Berg"):
    status, c = api("POST", "/contacts", who, {"bp_id": bp_id, "full_name": name, "email": email})
    assert status == 200, c
    return c


# ── contacts ───────────────────────────────────────────────────────────────────────────────

def test_one_email_is_one_contact():
    a = contact()
    assert contact(SOFIA, "ANNA@acme.com")["contact_id"] == a["contact_id"]        # same person, same company
    status, body = api("POST", "/contacts", TOM, {"bp_id": 999, "full_name": "Anna", "email": "anna@acme.com"})
    assert status == 409 and "already Anna Berg" in body["detail"]


def test_an_email_change_cannot_take_someone_elses():
    a, b = contact(), contact(email="bo@acme.com", name="Bo")
    status, body = api("PATCH", f"/contacts/{b['contact_id']}", TOM, {"email": "anna@acme.com"})
    assert status == 409 and "already belongs" in body["detail"]
    assert api("PATCH", f"/contacts/{b['contact_id']}", TOM, {"email": "bo.new@acme.com"})[0] == 200
    assert contact(email="bo@acme.com", name="Someone new")["contact_id"] != b["contact_id"]   # old one freed


def test_the_email_claim_is_a_plain_insert_so_a_racing_duplicate_fails(database):
    """Two adds that both pass the lookup: the second must fail at the database, not overwrite the first."""
    from brick_activity.infrastructure.tables import EmailTaken
    with use_cases.repository() as r:
        r.claim_email("race@acme.com", "C-FIRST")
    with pytest.raises(EmailTaken):
        with use_cases.repository() as r:
            r.claim_email("race@acme.com", "C-SECOND")
    with use_cases.repository() as r:
        from brick_activity.infrastructure.tables import emails
        from sqlalchemy import select
        assert r.conn.execute(select(emails.c.contact_id)).all() == [("C-FIRST",)]


def test_a_contact_who_left_cannot_be_subscribed():
    c = contact()
    api("POST", f"/contacts/{c['contact_id']}/left", TOM)
    status, body = api("PUT", f"/contacts/{c['contact_id']}/subscription", TOM, {"subscribed": True})
    assert status == 409 and "has left" in body["detail"]


def test_someone_elses_contact_is_theirs_to_change():
    c = contact(TOM)
    assert api("PATCH", f"/contacts/{c['contact_id']}", SOFIA, {"title": "CFO"})[0] == 403
    assert api("PATCH", f"/contacts/{c['contact_id']}", BOSS, {"title": "CFO"})[1]["title"] == "CFO"


# ── CRM-02 ─────────────────────────────────────────────────────────────────────────────────

def test_crm02_a_call_appears_on_the_contact_and_the_company_newest_first():
    c = contact()
    api("POST", "/activities", TOM, {"kind": "CALL", "subject": "first call", "contact_id": c["contact_id"],
                                     "occurred_at": "2026-10-01T09:00:00Z"})
    status, a = api("POST", "/activities", TOM, {"kind": "CALL", "subject": "Rang Anna",
                                                 "contact_id": c["contact_id"], "note": "proposal by Friday"})
    assert status == 200 and a["bp_id"] == str(ACME)                # the company came from the contact
    api("POST", "/activities", TOM, {"kind": "NOTE", "subject": "about Acme", "bp_id": ACME,
                                     "occurred_at": "2026-10-05T09:00:00Z"})
    on_contact = api("GET", "/timeline", READER, params={"contact_id": c["contact_id"]})[1]["activities"]
    on_company = api("GET", "/timeline", READER, params={"bp_id": ACME})[1]["activities"]
    assert [x["subject"] for x in on_contact] == ["Rang Anna", "first call"]
    assert [x["subject"] for x in on_company] == ["Rang Anna", "about Acme", "first call"]


def test_an_agent_logs_a_call_through_mcp():
    c = contact()
    verdict, a = anyio.run(_tool, TOM, "log_activity", {"kind": "CALL", "subject": "Rang Anna",
                                                       "contact_id": c["contact_id"]})
    assert verdict == "ok" and a["kind"] == "CALL"


def test_an_activity_must_be_about_someone_and_the_company_must_match():
    assert api("POST", "/activities", TOM, {"kind": "NOTE", "subject": "x"})[0] == 422
    c = contact()
    status, body = api("POST", "/activities", TOM, {"kind": "NOTE", "subject": "x", "contact_id": c["contact_id"],
                                                    "bp_id": 1})
    assert status == 409 and "works at partner 4242" in body["detail"]


def test_reading_is_shared_but_writing_needs_the_permission():
    assert api("POST", "/activities", READER, {"kind": "NOTE", "subject": "x", "bp_id": ACME})[0] == 403
    assert api("GET", "/timeline", READER, params={"bp_id": ACME})[0] == 200
    assert api("GET", "/timeline", {}, params={"bp_id": ACME})[0] == 401


# ── CRM-12 ─────────────────────────────────────────────────────────────────────────────────

def test_crm12_my_tasks_and_only_mine():
    for subject, due in (("late", "2026-10-10"), ("now", "2026-10-16"), ("soon", "2026-10-20")):
        api("POST", "/activities", TOM, {"kind": "TASK", "subject": subject, "bp_id": ACME, "due_on": due})
    api("POST", "/activities", TOM, {"kind": "TASK", "subject": "for sofia", "bp_id": ACME,
                                     "due_on": "2026-10-16", "assignee": "sofia@co.example"})
    board = api("GET", "/tasks/mine", TOM, params={"today": "2026-10-16"})[1]
    assert board["counts"] == {"overdue": 1, "today": 1, "this_week": 1}
    assert board["overdue"][0]["overdue"] is True
    assert [t["subject"] for t in api("GET", "/tasks/mine", SOFIA, params={"today": "2026-10-16"})[1]["today"]] \
        == ["for sofia"]


def test_only_the_assignee_or_a_manager_completes_a_task():
    t = api("POST", "/activities", TOM, {"kind": "TASK", "subject": "send", "bp_id": ACME, "due_on": "2026-10-20"})[1]
    assert api("POST", f"/tasks/{t['activity_id']}/complete", SOFIA, {})[0] == 403
    done = api("POST", f"/tasks/{t['activity_id']}/complete", TOM, {"note": "sent"})[1]
    assert done["status"] == "done" and done["note"] == "sent"
    assert api("PUT", f"/tasks/{t['activity_id']}/due", TOM, {"due_on": "2026-11-01"})[0] == 409


# ── tier 1: the real pipeline and business-partner bricks in this process ──────────────────

@pytest.fixture
def suite(monkeypatch):
    pytest.importorskip("brick_pipeline")
    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_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
    monkeypatch.setattr(bp_uc, "repository", sqlite_repo(bp_md, PartnerRepository))
    monkeypatch.setattr(p_uc, "repository", sqlite_repo(p_md, PipelineRepository))
    steward = {"sub": "s", "email": "s@co.example", "roles": [], "auth_source": "pool",
               "permissions": ["bp:partner:write"]}
    acme = bp_uc.upsert_partner(steward, "acme.com", "Acme AB", "SE", roles=["customer"])["bp_id"]
    for name, brick in (("business-partner", bp_app), ("pipeline", pipe_app)):
        register_local(name, brick)
    as_rep = lambda email: {"sub": email, "email": email, "roles": ["sales"], "auth_source": "pool",
                            "permissions": ["pipeline:opportunity:write"]}
    yield {"acme": acme, "p": p_uc, "rep": as_rep}
    for name in ("business-partner", "pipeline"):
        unregister_local(name)


def test_a_contact_must_work_at_a_real_partner(suite):
    status, body = api("POST", "/contacts", TOM, {"bp_id": 1, "full_name": "Ghost", "email": "g@x.com"})
    assert status == 422 and "no business partner 1" in body["detail"]
    assert contact(bp_id=suite["acme"])["bp_id"] == str(suite["acme"])


def test_crm02_the_call_lands_on_the_one_open_deal_and_keeps_it_from_going_quiet(suite):
    c = contact(bp_id=suite["acme"])
    deal = suite["p"].open_opportunity(suite["rep"]("tom@co.example"), suite["acme"], "Acme renewal", "EUR")
    status, a = api("POST", "/activities", TOM, {"kind": "CALL", "subject": "Rang Anna", "contact_id": c["contact_id"]})
    assert status == 200 and a["opportunity_id"] == deal["opportunity_id"] and a["notes"] == []
    on_deal = api("GET", "/timeline", TOM, params={"opportunity_id": deal["opportunity_id"]})[1]["activities"]
    assert [x["subject"] for x in on_deal] == ["Rang Anna"]
    events = suite["p"].opportunity_history(suite["rep"]("tom@co.example"), deal["opportunity_id"])["events"]
    assert events[-1]["action"] == "activity" and "Rang Anna" in events[-1]["detail"]


def test_with_two_open_deals_it_asks_rather_than_guesses(suite):
    c = contact(bp_id=suite["acme"])
    for name in ("Print", "Digital"):
        suite["p"].open_opportunity(suite["rep"]("tom@co.example"), suite["acme"], name, "EUR")
    a = api("POST", "/activities", TOM, {"kind": "CALL", "subject": "x", "contact_id": c["contact_id"]})[1]
    assert a["opportunity_id"] is None and "2 open deals" in a["notes"][0]


def test_another_reps_deal_is_not_where_my_call_lands(suite):
    c = contact(bp_id=suite["acme"])
    suite["p"].open_opportunity(suite["rep"]("sofia@co.example"), suite["acme"], "Sofia's deal", "EUR")
    a = api("POST", "/activities", TOM, {"kind": "CALL", "subject": "x", "contact_id": c["contact_id"]})[1]
    assert a["opportunity_id"] is None


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

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


def test_c4_a_bp_id_leaves_as_a_string_and_comes_in_as_either():
    c = contact(bp_id="45665172796086509")
    assert c["bp_id"] == "45665172796086509"
    assert api("GET", "/contacts", READER, params={"bp_id": "45665172796086509"})[1]["contacts"][0]["bp_id"] == \
        "45665172796086509"
    a = api("POST", "/activities", TOM, {"kind": "NOTE", "subject": "x", "bp_id": 45665172796086509})[1]
    assert a["bp_id"] == "45665172796086509"


def test_h12_a_bad_input_is_422_never_500():
    c = contact()
    for body in ({"kind": "NOTE", "subject": "nul\x00", "bp_id": ACME},
                 {"kind": "NOTE", "subject": "S" * 300, "bp_id": ACME},
                 {"kind": "TASK", "subject": "x", "bp_id": ACME, "due_on": "2026-02-30"},
                 {"kind": "TASK", "subject": "x", "bp_id": ACME, "due_on": "0001-01-01"},
                 {"kind": "CALL", "subject": "x", "bp_id": ACME, "occurred_at": "yesterday"},
                 {"kind": "NOTE", "subject": "x", "bp_id": "99999999999999999999999"}):
        status, out = api("POST", "/activities", TOM, body)
        assert status == 422, (body, status, out)
        assert tool(TOM, "log_activity", body)[0] == "refused"
    assert api("GET", "/tasks/mine", TOM, params={"today": "not-a-date"})[0] == 422
    assert tool(TOM, "my_tasks", {"today": "tomorrow"})[1].startswith("422") or \
        "today" in tool(TOM, "my_tasks", {"today": "tomorrow"})[1]
    assert api("GET", "/timeline", TOM, params={"contact_id": c["contact_id"], "limit": -1})[0] == 422
    assert api("POST", "/contacts", TOM, {"bp_id": ACME, "full_name": "N" * 300, "email": "n@acme.com"})[0] == 422


def test_h12_two_adds_of_one_email_at_once_give_the_same_contact_not_a_500(monkeypatch):
    """What DataK3 does to the second of two concurrent adds: its lookup saw no contact (its snapshot), and its
    email claim fails — at the INSERT, or only at COMMIT. Either way it gets the contact the first one made."""
    from brick_activity.infrastructure.tables import ActivityRepository
    first = contact()
    real_lookup = ActivityRepository.contact_by_email
    blind = {"n": 1}

    def stale(self, email):
        if blind["n"]:
            blind["n"] -= 1
            return None
        return real_lookup(self, email)
    monkeypatch.setattr(ActivityRepository, "contact_by_email", stale)
    assert contact(SOFIA, "Anna@acme.com", name="Anna B")["contact_id"] == first["contact_id"]

    from sqlalchemy.exc import IntegrityError
    blind["n"] = 1
    real_claim = ActivityRepository.claim_email
    monkeypatch.setattr(ActivityRepository, "claim_email", lambda self, e, c: None)          # the claim "passes"…
    real_save = ActivityRepository.save_contact

    def save_then_lose_at_commit(self, c):
        real_save(self, c)
        raise IntegrityError("COMMIT", {}, Exception('duplicate key value violates unique constraint '
                                                     '"activity_contact_emails_pkey" (SQLSTATE 23505)'))
    monkeypatch.setattr(ActivityRepository, "save_contact", save_then_lose_at_commit)
    status, again = api("POST", "/contacts", TOM, {"bp_id": ACME, "full_name": "Anna", "email": "anna@acme.com"})
    assert status == 200 and again["contact_id"] == first["contact_id"], again
    monkeypatch.setattr(ActivityRepository, "claim_email", real_claim)
    monkeypatch.setattr(ActivityRepository, "save_contact", real_save)
    blind["n"] = 1
    status, other = api("POST", "/contacts", TOM, {"bp_id": 999, "full_name": "Anna", "email": "anna@acme.com"})
    assert status == 409 and "already Anna Berg" in other["detail"]                    # another company: refused


def test_c3_a_contact_and_a_task_carry_versions_and_a_stale_one_is_409():
    c = contact()
    assert c["version"] == 1
    edited = api("PATCH", f"/contacts/{c['contact_id']}", TOM, {"title": "CFO", "version": 1})[1]
    assert edited["version"] == 2
    assert api("PATCH", f"/contacts/{c['contact_id']}", TOM, {"title": "CEO", "version": 1}) == \
        (409, {"detail": "changed since you opened it — reload it and make your change again"})
    t = api("POST", "/activities", TOM, {"kind": "TASK", "subject": "send", "bp_id": ACME, "due_on": "2026-10-20"})[1]
    assert t["version"] == 1
    assert api("PUT", f"/tasks/{t['activity_id']}/due", TOM, {"due_on": "2026-10-22", "version": 1})[0] == 200
    assert api("POST", f"/tasks/{t['activity_id']}/complete", TOM, {"version": 1})[0] == 409
    assert tool(TOM, "complete_task", {"activity_id": t["activity_id"], "version": 1})[1].startswith("409")
    assert api("POST", f"/tasks/{t['activity_id']}/complete", TOM, {"version": 2})[1]["status"] == "done"


def test_c3_two_edits_of_one_contact_without_a_version_both_land(monkeypatch):
    from brick_activity.infrastructure.tables import ActivityRepository
    c = contact()
    with use_cases.repository() as r:
        snapshot = r.contact(c["contact_id"])
    api("PATCH", f"/contacts/{c['contact_id']}", TOM, {"title": "CFO"})          # the other edit commits first
    real = ActivityRepository.contact
    once = {"n": 1}

    def stale(self, contact_id):
        if once["n"] and contact_id == c["contact_id"]:
            once["n"] -= 1
            import copy
            return copy.deepcopy(snapshot)
        return real(self, contact_id)
    monkeypatch.setattr(ActivityRepository, "contact", stale)
    assert api("PATCH", f"/contacts/{c['contact_id']}", TOM, {"phone": "+46 8 123"})[0] == 200
    now = api("GET", f"/contacts/{c['contact_id']}", READER)[1]
    assert (now["title"], now["phone"], now["version"]) == ("CFO", "+46 8 123", 3)


def test_h13_a_retried_log_with_one_idempotency_key_is_one_activity(tmp_path, monkeypatch):
    from sqlalchemy import create_engine as ce
    from erp_platform import idempotency
    e = ce(f"sqlite:///{tmp_path / 'idem.db'}")
    idempotency.md.create_all(e)
    monkeypatch.setattr(use_cases.ACT, "idempotency_connection", e.begin)
    body = {"kind": "CALL", "subject": "Rang Anna", "bp_id": ACME}
    r1 = anyio.run(_api_h, "POST", "/activities", {**TOM, "Idempotency-Key": "call-1"}, body)
    r2 = anyio.run(_api_h, "POST", "/activities", {**TOM, "Idempotency-Key": "call-1"}, body)
    assert r1.status_code == r2.status_code == 200 and r1.json() == r2.json()
    assert r2.headers["idempotent-replayed"] == "true"
    one = tool(TOM, "log_activity", {**body, "idempotency_key": "call-2"})
    two = tool(TOM, "log_activity", {**body, "idempotency_key": "call-2"})
    assert one == two
    on_company = api("GET", "/timeline", READER, params={"bp_id": ACME})[1]["activities"]
    assert len(on_company) == 2


async def _api_h(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_h5_a_deals_timeline_is_for_those_who_may_see_the_deal(suite):
    c = contact(bp_id=suite["acme"])
    deal = suite["p"].open_opportunity(suite["rep"]("tom@co.example"), suite["acme"], "Acme renewal", "EUR")
    api("POST", "/activities", TOM, {"kind": "CALL", "subject": "Rang Anna", "contact_id": c["contact_id"],
                                     "opportunity_id": deal["opportunity_id"]})
    oid = deal["opportunity_id"]
    assert len(api("GET", "/timeline", TOM, params={"opportunity_id": oid})[1]["activities"]) == 1
    assert len(api("GET", "/timeline", BOSS, params={"opportunity_id": oid})[1]["activities"]) == 1
    assert api("GET", "/timeline", SOFIA, params={"opportunity_id": oid}) == (404, {"detail": f"no opportunity {oid}"})
    assert tool(SOFIA, "timeline", {"opportunity_id": oid}) == ("refused", f"404: no opportunity {oid}")
    assert api("GET", "/timeline", TOM, params={"opportunity_id": "O-NOPE"})[0] == 404
    # and a call cannot be logged onto a deal its rep cannot see
    status, out = api("POST", "/activities", SOFIA, {"kind": "NOTE", "subject": "x", "bp_id": suite["acme"],
                                                     "opportunity_id": oid})
    assert (status, out) == (404, {"detail": f"no opportunity {oid}"})


def test_h5_a_contact_or_company_timeline_that_does_not_exist_is_404(suite):
    assert api("GET", "/timeline", TOM, params={"contact_id": "C-NOPE"}) == (404, {"detail": "no contact C-NOPE"})
    assert api("GET", "/timeline", TOM, params={"bp_id": 123})[0] == 404
    assert api("GET", "/contacts", TOM, params={"bp_id": 123})[0] == 404
    assert api("GET", "/timeline", TOM, params={"bp_id": suite["acme"]})[0] == 200


# ── 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_activity/roles.yaml — the roles this brick ships, from its actors. They hold exactly the permissions the brick
    decides by (its use cases' and CONTACTS_MANAGE, TASKS_MANAGE), and a machine permission only on a service client."""
    import pathlib
    import yaml
    from brick_activity.application.use_cases import ACT, CONTACTS_MANAGE, TASKS_MANAGE
    spec = yaml.safe_load((pathlib.Path(use_cases.__file__).parents[1] / "roles.yaml").read_text())
    known = {uc.permission for uc in ACT.use_cases if uc.permission} | {CONTACTS_MANAGE, TASKS_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(ACT.machine_permissions), role


def test_a_role_name_decides_nothing():
    """The `manager` role name without the permission changes only its own contacts; any role holding
    activity:contact:manage changes another's."""
    c = contact()
    named = person("nam@co.example", ["manager"], REP)
    assert api("PATCH", f"/contacts/{c['contact_id']}", named, {"title": "x"})[0] == 403
    changer = person("lead@co.example", ["team_lead"], REP + ["activity:contact:manage"])
    assert api("PATCH", f"/contacts/{c['contact_id']}", changer, {"title": "x"})[1]["title"] == "x"


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 ("log_task", "complete_task", "cancel_task", "reopen_task"):
        assert f"Task_{key}" in text, key