dodil/erp-brick-business-partner

tests/test_doors.py

"""The partner use cases through both doors over the REAL repository on SQLite. scripts/live_check.py
runs the same story on DataK3."""
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_bp.app import app
from brick_bp.application import use_cases
from brick_bp.domain.partner import bp_id_for
from brick_bp.infrastructure.tables import PartnerRepository, 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 PartnerRepository(conn)
    monkeypatch.setattr(use_cases, "repository", repo)
    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"}


STEWARD = person("sam@co.example", ["steward"], ["bp:partner:write", "bp:family:write"])
SALES_APP = person("crm@co.example", ["service"], ["bp:partner:promote"])
ANYONE = person("ann@co.example", ["sales"], [])


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 tool(who, name, args):
    return anyio.run(_tool, who, name, args)


def make(key, name, country="SE", **kw):
    status, p = api("PUT", "/partners", STEWARD, {"bp_key": key, "legal_name": name, "country": country, **kw})
    assert status == 200, p
    return p


# ── BP-01 / BP-02 ──────────────────────────────────────────────────────────────────────────

def test_bp02_importing_the_same_company_twice_creates_it_once():
    a = make("acme.com", "Acme AB", roles=["customer"])
    b = make("ACME.com", "Acme AB", roles=["customer"])
    assert a["bp_id"] == b["bp_id"] == str(bp_id_for("acme.com"))
    assert len(api("GET", "/partners", ANYONE, params={"name": "acme"})[1]["partners"]) == 1
    events = api("GET", f"/partners/{a['bp_id']}/history", ANYONE)[1]["events"]
    assert [e["action"] for e in events] == ["created"]                # the re-import changed nothing


def test_bp01_sales_and_invoicing_mean_the_same_customer():
    make("acme.com", "Acme AB", roles=["customer"])
    p = make("acme.com", "Acme AB", roles=["supplier"])
    assert p["roles"] == ["customer", "supplier"]


def test_a_reimport_never_unblocks_a_blocked_customer():
    """The reference package defaulted status to active on every write."""
    p = make("acme.com", "Acme AB")
    api("PUT", f"/partners/{p['bp_id']}/status", STEWARD, {"status": "blocked", "reason": "unpaid"})
    again = make("acme.com", "Acme AB")
    assert again["status"] == "blocked" and again["can_transact"] is False


def test_a_reimport_without_a_tax_id_keeps_the_one_on_file():
    p = make("acme.com", "Acme AB", tax_id="SE5560123401")
    assert make("acme.com", "Acme AB")["tax_id"] == "SE5560123401"
    assert p["tax_id"] == "SE5560123401"


# ── BP-03: the family, typed ───────────────────────────────────────────────────────────────

def test_bp03_the_family_walk_is_typed():
    group = make("acme-group", "Acme Group", "GB")["bp_id"]
    uk = make("acme-uk", "Acme UK", "GB")["bp_id"]
    de = make("acme-de", "Acme DE", "DE")["bp_id"]
    berlin = make("acme-berlin", "Acme DE Berlin", "DE")["bp_id"]
    other = make("frame-co", "Framework Counterparty", "FR")["bp_id"]
    for child, parent, rel in ((uk, group, "subsidiary_of"), (de, group, "subsidiary_of"),
                               (berlin, de, "subsidiary_of"), (other, group, "framework_governs")):
        assert api("PUT", f"/partners/{child}/parents", STEWARD, {"parent_id": parent, "rel": rel})[0] == 200
    fam = api("GET", f"/partners/{group}/family", ANYONE)[1]
    assert {m["legal_name"] for m in fam["below"]} == {"Acme UK", "Acme DE", "Acme DE Berlin"}
    assert other not in [m["bp_id"] for m in fam["below"]]              # a framework does not leak in
    # and an agent asks the same question through MCP
    verdict, via_mcp = tool(ANYONE, "get_family", {"bp_id": group})
    assert verdict == "ok" and via_mcp["below"] == fam["below"]


def test_a_cycle_and_a_second_parent_are_refused():
    a, b, c = (make(k, n)["bp_id"] for k, n in (("a", "A AB"), ("b", "B AB"), ("c", "C AB")))
    api("PUT", f"/partners/{b}/parents", STEWARD, {"parent_id": a})
    api("PUT", f"/partners/{c}/parents", STEWARD, {"parent_id": b})
    status, body = api("PUT", f"/partners/{a}/parents", STEWARD, {"parent_id": c})
    assert status == 409 and "cycle" in body["detail"]
    status, body = api("PUT", f"/partners/{c}/parents", STEWARD, {"parent_id": a})
    assert status == 409 and "one parent" in body["detail"]


def test_linking_an_unknown_partner_is_not_found():
    a = make("a", "A AB")["bp_id"]
    assert api("PUT", f"/partners/{a}/parents", STEWARD, {"parent_id": 12345})[0] == 404


# ── BP-04: the seam other modules call ─────────────────────────────────────────────────────

def test_bp04_promote_creates_or_reuses_and_adds_the_role():
    status, p = api("POST", "/partners/promote", SALES_APP,
                    {"bp_key": "newco.example", "legal_name": "NewCo Ltd", "country": "GB"})
    assert status == 200 and p["roles"] == ["customer"]
    status, again = api("POST", "/partners/promote", STEWARD_PLUS, {"bp_key": "newco.example", "role": "supplier"})
    assert again["bp_id"] == p["bp_id"] and again["roles"] == ["customer", "supplier"]
    assert again["legal_name"] == "NewCo Ltd"                          # details are the steward's to change


def test_bp04_promoting_an_unknown_party_without_a_name_says_what_is_missing():
    status, body = api("POST", "/partners/promote", SALES_APP, {"bp_key": "ghost.example"})
    assert status == 422 and "legal_name" in body["detail"]


# ── access ─────────────────────────────────────────────────────────────────────────────────

def test_without_bp_partner_write_nothing_is_created_or_edited():
    assert api("PUT", "/partners", ANYONE, {"bp_key": "x", "legal_name": "X", "country": "SE"})[0] == 403
    assert api("PUT", "/partners", SALES_APP, {"bp_key": "x", "legal_name": "X", "country": "SE"})[0] == 403
    assert tool(ANYONE, "upsert_partner", {"bp_key": "x", "legal_name": "X", "country": "SE"}) == \
        ("refused", "403: missing permission: bp:partner:write")
    assert api("GET", "/partners", ANYONE, params={"bp_key": "x"})[1] == {"partners": []}


def test_anyone_signed_in_reads_partners_and_nobody_unsigned_does():
    p = make("acme.com", "Acme AB")
    assert api("GET", f"/partners/{p['bp_id']}", ANYONE)[0] == 200
    assert api("GET", f"/partners/{p['bp_id']}", {})[0] == 401


def test_a_status_change_needs_a_reason():
    p = make("acme.com", "Acme AB")
    status, body = api("PUT", f"/partners/{p['bp_id']}/status", STEWARD, {"status": "blocked", "reason": " "})
    assert status == 422


# ── person partners (BP-05..BP-07) ─────────────────────────────────────────────────────────

HR = person("hr-service@co.example", ["service"], ["bp:person:sync", "bp:person:read"])
CONTRACTS = person("legal@co.example", ["legal"], ["bp:person:read"])
STEWARD_PLUS = person("sam@co.example", ["steward"], ["bp:partner:write", "bp:family:write", "bp:partner:promote"])
CARD = {"person_id": "812", "legal_name": "Anna Berg", "preferred_name": "Anna",
        "work_email": "anna@co.example", "work_phone": "+44 20 7946 0000", "roles": ["employee"]}


def sync(card=None, who=HR, anonymise=False):
    return api("POST", "/partners/persons/sync", who, {"person": card or CARD, "anonymise": anonymise})


def test_bp05_hr_syncs_a_person_once_and_a_second_identical_call_changes_nothing():
    status, p = sync()
    assert status == 200 and p["kind"] == "person" and p["bp_id"] == str(bp_id_for("person:812"))
    assert (p["legal_name"], p["work_email"], p["roles"]) == ("Anna Berg", "anna@co.example", ["employee"])
    again = sync()
    assert again == (200, p)
    events = api("GET", f"/partners/{p['bp_id']}/history", HR)[1]["events"]
    assert [e["action"] for e in events] == ["person created"]
    assert "Anna" not in repr(events) and "anna@co.example" not in repr(events)     # names, never values
    verdict, via_mcp = tool(HR, "sync_person", {"person": CARD})
    assert verdict == "ok" and via_mcp == p


def test_bp05_a_private_field_is_refused_through_both_doors():
    status, body = sync({**CARD, "personal_email": "anna@home.example"})
    assert status == 422 and "a person partner carries a name and a work contact only" in body["detail"]
    verdict, reason = tool(HR, "sync_person", {"person": {**CARD, "home_address": "1 Road"}})
    assert verdict == "refused" and reason.startswith("422: a person partner carries a name and a work contact only")
    assert api("GET", "/partners", HR, params={"bp_key": "person:812"})[1] == {"partners": []}   # nothing written


def test_bp05_works_for_links_a_freelancer_to_their_company():
    psc = make("anna-consulting.example", "Anna Consulting Ltd", "GB", roles=["supplier"])
    status, p = sync({**CARD, "roles": ["worker"], "works_for": psc["bp_id"]})
    assert status == 200
    fam = api("GET", f"/partners/{psc['bp_id']}/family", CONTRACTS, params={"rel": "works_for"})[1]
    assert [m["bp_id"] for m in fam["below"]] == [p["bp_id"]]
    hidden = api("GET", f"/partners/{psc['bp_id']}/family", ANYONE, params={"rel": "works_for"})[1]
    assert hidden["below"] == []                                        # people are not shown to everyone
    sync({**CARD, "roles": ["employee"], "works_for": None})           # hired: no longer through the company
    assert api("GET", f"/partners/{psc['bp_id']}/family", CONTRACTS, params={"rel": "works_for"})[1]["below"] == []
    assert sync({**CARD, "works_for": 4242})[0] == 422


def test_bp05_partner_write_cannot_touch_a_person():
    p = sync()[1]
    for method, path, body in (("PUT", "/partners", {"bp_key": "person:812", "legal_name": "X", "country": "GB"}),
                               ("PUT", f"/partners/{p['bp_id']}/status", {"status": "blocked", "reason": "x"}),
                               ("DELETE", f"/partners/{p['bp_id']}/roles/employee", None),
                               ("POST", "/partners/promote", {"bp_key": "person:812", "role": "customer"})):
        status, out = api(method, path, STEWARD_PLUS, body)
        assert status == 404 or (status == 409 and "people are written by hr" in out["detail"]), (path, status, out)
    # the steward cannot see people, so a write by id is 404 — it does not even confirm the person exists
    assert api("PUT", f"/partners/{p['bp_id']}/status", STEWARD_PLUS, {"status": "blocked", "reason": "x"})[0] == 404
    assert api("GET", "/partners", HR, params={"bp_key": "person:812"})[1]["partners"][0]["legal_name"] == "Anna Berg"
    assert sync(who=STEWARD)[0] == 403                                  # bp:partner:write is not bp:person:sync


def test_bp06_without_person_read_a_person_is_absent_and_404_never_403():
    acme = make("acme.com", "Acme AB", roles=["customer"])
    p = sync({**CARD, "legal_name": "Acme Starter"})[1]
    listed = api("GET", "/partners", ANYONE, params={"name": "acme"})[1]["partners"]
    assert [x["bp_id"] for x in listed] == [acme["bp_id"]]
    assert api("GET", f"/partners/{p['bp_id']}", ANYONE)[0] == 404
    assert api("GET", f"/partners/{p['bp_id']}/history", ANYONE)[0] == 404
    assert api("GET", "/partners", ANYONE, params={"bp_key": "person:812"})[1] == {"partners": []}
    assert tool(ANYONE, "get_partner", {"bp_id": p["bp_id"]})[1].startswith("404")
    both = api("GET", "/partners", CONTRACTS, params={"name": "acme"})[1]["partners"]
    assert {x["bp_id"] for x in both} == {acme["bp_id"], p["bp_id"]}
    assert api("GET", f"/partners/{acme['bp_id']}", ANYONE)[1]["kind"] == "organisation"   # organisations as before


def test_bp07_anonymise_erases_the_person_and_keeps_the_id():
    p = sync()[1]
    status, gone = sync({"person_id": "812"}, anonymise=True)
    assert status == 200 and gone["bp_id"] == p["bp_id"] and gone["legal_name"] == "Erased person"
    assert gone["work_email"] == "" and gone["roles"] == ["employee"] and gone["anonymised"] is True
    assert sync({"person_id": "812"}, anonymise=True) == (200, gone)    # a second call changes nothing
    status, body = sync()
    assert status == 409 and "erased" in body["detail"]
    history = api("GET", f"/partners/{p['bp_id']}/history", HR)[1]["events"]
    assert [e["action"] for e in history] == ["person created", "person anonymised"]
    assert "Anna" not in repr(history)


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

SALES = person("tom@co.example", ["sales"], ["bp:partner:promote"])
BIG = 45665172796086509                      # a bp_id past 2**53: a JavaScript number reads it as …512


def test_c4_a_bp_id_leaves_as_a_string_and_comes_in_as_either(database):
    p = make("acme.com", "Acme AB", roles=["customer"])
    assert p["bp_id"] == str(bp_id_for("acme.com"))
    assert api("GET", f"/partners/{p['bp_id']}", ANYONE)[1]["bp_id"] == p["bp_id"]
    assert tool(ANYONE, "get_partner", {"bp_id": p["bp_id"]}) == ("ok", api("GET", f"/partners/{p['bp_id']}", ANYONE)[1])
    assert tool(ANYONE, "get_partner", {"bp_id": int(p["bp_id"])})[0] == "ok"
    parent = make("acme-group", "Acme Group", "GB")
    linked = api("PUT", f"/partners/{p['bp_id']}/parents", STEWARD, {"parent_id": parent["bp_id"]})[1]
    assert linked["parents"] == [parent["bp_id"]] and linked["bp_id"] == p["bp_id"]
    fam = api("GET", f"/partners/{parent['bp_id']}/family", ANYONE)[1]
    assert [m["bp_id"] for m in fam["below"]] == [p["bp_id"]]


def test_h12_a_bad_input_is_422_never_500():
    for body in ({"bp_key": "nul.example", "legal_name": "Nul\x00byte", "country": "SE"},
                 {"bp_key": "long.example", "legal_name": "L" * 5000, "country": "SE"},
                 {"bp_key": "k" * 3000, "legal_name": "Long key", "country": "SE"}):
        status, out = api("PUT", "/partners", STEWARD, body)
        assert status == 422, (status, out)
        assert tool(STEWARD, "upsert_partner", body)[0] == "refused"
    assert api("GET", "/partners/99999999999999999999999", ANYONE)[0] == 422
    assert api("GET", "/partners", ANYONE, params={"limit": -1})[0] == 422


def test_h14_sales_may_promote_a_customer_only():
    status, p = api("POST", "/partners/promote", SALES,
                    {"bp_key": "newco.example", "legal_name": "NewCo", "country": "GB"})
    assert status == 200 and p["roles"] == ["customer"]
    for role in ("supplier", "carrier"):
        assert api("POST", "/partners/promote", SALES, {"bp_key": "newco.example", "role": role}) == \
            (403, {"detail": "missing permission: bp:partner:write"})
        assert tool(SALES, "promote_partner", {"bp_key": "newco.example", "role": role}) == \
            ("refused", "403: missing permission: bp:partner:write")
    assert api("GET", f"/partners/{p['bp_id']}", ANYONE)[1]["roles"] == ["customer"]
    assert api("POST", "/partners/promote", STEWARD_PLUS, {"bp_key": "newco.example", "role": "supplier"})[1][
        "roles"] == ["customer", "supplier"]


def test_c3_the_partner_carries_a_version_and_a_stale_one_is_409():
    p = make("acme.com", "Acme AB")
    assert p["version"] == 1
    status, blocked = api("PUT", f"/partners/{p['bp_id']}/status", STEWARD,
                          {"status": "blocked", "reason": "unpaid", "version": 1})
    assert status == 200 and blocked["version"] == 2
    stale = api("PUT", f"/partners/{p['bp_id']}/status", STEWARD,
                {"status": "active", "reason": "paid", "version": 1})
    assert stale == (409, {"detail": "changed since you opened it — reload it and make your change again"})
    assert tool(STEWARD, "set_partner_status", {"bp_id": p["bp_id"], "status": "active", "reason": "paid",
                                                "version": 1})[1].startswith("409: changed since you opened it")
    assert api("GET", f"/partners/{p['bp_id']}", ANYONE)[1]["status"] == "blocked"


def test_c3_if_match_carries_the_version_and_the_get_sends_it_as_the_etag(database):
    p = make("acme.com", "Acme AB")
    r = anyio.run(_api_raw, "GET", f"/partners/{p['bp_id']}", ANYONE, None, {})
    assert r.headers["etag"] == '"1"'
    r = anyio.run(_api_raw, "PUT", f"/partners/{p['bp_id']}/status", STEWARD,
                  {"status": "blocked", "reason": "x"}, {"If-Match": '"1"'})
    assert r.status_code == 200
    r = anyio.run(_api_raw, "PUT", f"/partners/{p['bp_id']}/status", STEWARD,
                  {"status": "active", "reason": "x"}, {"If-Match": '"1"'})
    assert r.status_code == 409


def test_c3_two_stewards_without_a_version_both_land(database, monkeypatch):
    """The second of two concurrent saves read the partner before the first committed (what DataK3's snapshot
    gives it). Without a version it is re-run on the latest: the role AND the status both stick."""
    p = make("acme.com", "Acme AB")
    stale = stale_once(monkeypatch, database, int(p["bp_id"]))
    api("PUT", "/partners", STEWARD, {"bp_key": "acme.com", "legal_name": "Acme AB", "country": "SE",
                                      "roles": ["supplier"]})
    stale.arm()
    assert api("PUT", f"/partners/{p['bp_id']}/status", STEWARD, {"status": "blocked", "reason": "x"})[0] == 200
    now = api("GET", f"/partners/{p['bp_id']}", ANYONE)[1]
    assert now["roles"] == ["supplier"] and now["status"] == "blocked" and now["version"] == 3


async def _api_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 stale_once(monkeypatch, engine, bp_id):
    """The next load of `bp_id` returns the partner as it is NOW — armed later, after another save committed."""
    with engine.connect() as c:
        snapshot = PartnerRepository(c).get(bp_id)
    real = PartnerRepository.get

    class Stale:
        armed = False

        def arm(self):
            self.armed = True

    s = Stale()

    def get(self, i):
        if s.armed and int(i) == bp_id:
            s.armed = False
            import copy
            return copy.deepcopy(snapshot)
        return real(self, i)
    monkeypatch.setattr(PartnerRepository, "get", get)
    return s


def test_later_bricks_read_the_odoo_map_through_one_route(database):
    """shared.yaml conventions.odoo_migration: GET /migration-map/{db}/{model}?ids= -> [{odoo_model, odoo_id, ours}]."""
    p = make("acme", "Acme AB")
    with database.begin() as conn:
        PartnerRepository(conn).map_put("odoo17", "res.partner", 42, int(p["bp_id"]))
    s, got = api("GET", "/migration-map/odoo17/res.partner", ANYONE, params={"ids": "42,43"})
    assert s == 200 and got == [{"odoo_model": "res.partner", "odoo_id": "42", "ours": str(p["bp_id"])}]