dodil/erp-brick-expenses

DODIL ERP brick: expenses

tests/test_doors.py

"""Both doors over the real app: the same use case answers the same through the API and through MCP (the agent door), a claim of another
person is a 404, a missing permission a 403 naming it, no identity a 401, and the routes only another module calls are not tools."""
import anyio
import httpx
from conftest import EMPLOYEE, FINANCE, MANAGER, NOBODY, api, app, claim_551, new_report, refused, submit, tool, tools
from erp_platform import seam

TOOLS = ["add_line", "add_mileage", "add_per_diem", "add_receipt", "ai_consent_status", "approve_level", "cancel_payout", "cancel_report",
         "create_report", "delegate_approval", "discard_exp_migration", "edit_line", "end_delegation", "explain_exp_check",
         "get_exp_migration_map", "get_exp_migration_run", "get_receipt", "get_report", "import_card_feed", "list_card_txns",
         "list_categories", "list_delegations", "list_exp_migration_runs", "list_policy_rules", "list_rates", "list_reminders",
         "list_reports", "list_settings", "load_exp_from_odoo", "mark_personal", "mark_rebillable", "match_card_txn", "post_report",
         "put_card", "put_category", "put_policy_rule", "put_rate", "recompute_report", "reconcile_exp_migration", "recover_personal",
         "reissue_payout", "reject_report", "remove_line", "return_report", "set_ai_consent", "set_setting", "sign_exp_reconciliation",
         "submit_report", "suggest_card_matches"]


def test_the_agent_door_offers_the_use_cases_and_never_the_seam_only_routes():
    assert tools(EMPLOYEE) == TOOLS
    for seam_only in ("confirm_paid", "intake_leaver", "drain_outbox", "show_process_bpmn", "run_reminders"):
        assert seam_only not in tools(EMPLOYEE)


def test_a_claim_is_built_submitted_and_approved_through_the_agent_door_alone(nb):
    status, r = tool(EMPLOYEE, "create_report", {"currency": "GBP", "title": "Agent claim"})
    assert status == "ok" and r["status"] == "draft"
    rid = r["report_id"]
    status, r = tool(EMPLOYEE, "add_line", {"report_id": rid, "category_code": "TRAVEL_RAIL", "expense_date": "2026-11-04", "merchant": "Northern Rail",
                                            "amount": "86.40", "receipt_key": f"raw/expenses/{rid}/1/{'a' * 64}.jpg"})
    assert status == "ok" and r["total_amount"] == "86.40"
    assert tool(EMPLOYEE, "submit_report", {"report_id": rid})[1]["chain"][0]["kind"] == "manager"
    assert tool(MANAGER, "list_reports", {"scope": "waiting"})[1]["reports"][0]["report_id"] == rid
    assert tool(MANAGER, "approve_level", {"report_id": rid})[1]["status"] == "posted"
    assert tool(EMPLOYEE, "get_report", {"report_id": rid})[1]["journal_id"].isdigit()


def test_the_two_doors_agree_on_every_refusal(nb):
    rid = claim_551()
    other = new_report()["report_id"]
    cases = [("submit_report", {"report_id": rid}, "POST", f"/reports/{rid}/submit", None)]
    submit(rid)
    cases += [("submit_report", {"report_id": rid}, "POST", f"/reports/{rid}/submit", None),
              ("approve_level", {"report_id": rid}, "POST", f"/reports/{rid}/approve", None),
              ("return_report", {"report_id": rid, "reason": "x"}, "POST", f"/reports/{rid}/return", None)]
    for name, args, method, path, _ in cases[1:]:
        who = EMPLOYEE if name == "submit_report" else FINANCE
        status, text = tool(who, name, args)
        code, body = api(method, path, who, args if method == "POST" and len(args) > 1 else None)
        assert status == "refused" and code in (403, 409), (name, code, body)
        assert body["detail"] in text or text in body["detail"], (name, text, body)
    assert tool(MANAGER, "get_report", {"report_id": other})[0] == "refused"        # a draft of another person: not found, on both doors
    refused(api("GET", f"/reports/{other}", MANAGER), 404)
    assert tool(FINANCE, "get_report", {"report_id": other})[0] == "ok"             # finance reads every claim (exp:report:read_all)


def test_no_identity_is_a_401_and_no_permission_a_403_naming_it():
    async def bare():
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://brick") as c:
            return [(await c.get("/reports")).status_code, (await c.post("/reports", json={"currency": "GBP"})).status_code]
    assert anyio.run(bare) == [401, 401]
    refused(api("POST", "/reports", NOBODY, {"currency": "GBP"}), 403, "exp:report:write_own")
    refused(api("GET", "/reports", NOBODY), 403, "exp:report:read_own")
    refused(api("POST", "/outbox/drain", EMPLOYEE), 403, "exp:journal:post_auto")


def test_the_mcp_door_refuses_what_the_api_refuses_for_a_lost_hr(nb):
    seam.unregister_local("hr")
    status, text = tool(EMPLOYEE, "create_report", {"currency": "GBP"})
    assert status == "refused" and "hr is not part of this installation" in text