dodil/erp-brick-ar

DODIL ERP brick: accounts receivable

tests/test_access.py

"""Who may do what: the permission matrix over every use case, the machine permissions, the controls, the second person, and the
narrow route sales gets. A brick decides by PERMISSION; the personas carry the permissions of a role and never its name."""
import pathlib
import re
from decimal import Decimal

import pytest
import yaml
from conftest import (
    APPROVER,
    BILLER,
    BP,
    COLLECTOR,
    NOBODY,
    OWNER,
    ROLES,
    SALES,
    SYSTEM,
    api,
    billing,
    control,
    hand_draft,
    issued,
    line,
    must,
    person,
    tool,
    tools,
)

from brick_ar.application.services import MACHINE
from brick_ar.application.use_cases import AR

USE_CASES = [u for u in AR.use_cases if u.permission]          # show_process_bpmn is the platform's open read of the signed process
FILL = {"invoice_id": "x", "credit_id": "x", "writeoff_id": "x", "schedule_id": "x", "bp_id": "1", "run_id": "x", "run_no": "1", "step": "ai_check_draft",
        "source_kind": "sales_order", "source_id": "x", "document_id": "x", "batch_id": "x", "request_id": "x"}
PEOPLE = {"biller": BILLER, "collector": COLLECTOR, "approver": APPROVER, "owner": OWNER, "sales": SALES}


def dummy(name, ann):
    """A minimal VALID value for a required input (the platform validates before it checks the permission)."""
    import typing
    origin = typing.get_origin(ann)
    if origin is typing.Annotated:
        ann = typing.get_args(ann)[0]
        origin = typing.get_origin(ann)
    if origin is typing.Literal:
        return typing.get_args(ann)[0]
    if origin is list:
        return [{"description": "x", "quantity": "1", "unit_price": "1.00"}]
    named = {"period": "2026-10", "starts_on": "2026-11-20", "cutover_date": "2026-11-20", "run_no": 1, "line_no": 1, "periods": 1,
             "days": 30, "value": True, "account_id": "1", "bp_id": "1", "quantity": "1", "unit_price": "1.00", "amount": "1.00", "reason": "r",
             "received_on": "2026-11-20", "currency": "GBP", "discount_amount": "1.00", "net_amount": "1.00", "valid_from": "2026-06-30",
             "reference_rate_pct": "4.25", "rate_pct": "5", "period_id": "2026-10", "bucket": "0-30"}
    if name in named:
        return named[name]
    return {int: 1, bool: True}.get(ann, "x")


def path_of(uc):
    return re.sub(r"\{(\w+)\}", lambda m: str(dummy(m.group(1), str) if m.group(1) not in FILL else FILL[m.group(1)]), uc.path)


def call(uc, headers):
    import inspect
    body = {n: dummy(n, a) for n, a, d in uc.inputs if d is inspect.Parameter.empty and n not in uc.path_params}
    return api(uc.method, path_of(uc), headers, body if body or uc.method != "GET" else None)[0]


def test_every_use_case_declares_a_concrete_permission_and_the_machine_ones_are_exactly_the_spec_list():
    assert all(uc.permission and uc.permission.count(":") == 2 and "*" not in uc.permission for uc in USE_CASES)
    from conftest import METHOD_DIR
    spec = yaml.safe_load((pathlib.Path(METHOD_DIR) / "modules" / "ar" / "module.yaml").read_text())
    assert set(MACHINE) == set(spec["permissions"]["machine"]) and set(USE_CASES and AR.machine_permissions) == set(MACHINE)
    spec_catalog = set(spec["permissions"]["catalog"])
    assert {uc.permission for uc in USE_CASES} <= spec_catalog                       # no permission the spec does not publish
    assert set(ROLES["system"]) - set(MACHINE) == {"ar:invoice:read"} and spec_catalog >= {p for r in ROLES.values() for p in r}


@pytest.mark.parametrize("uc", USE_CASES, ids=lambda u: u.name)
def test_the_permission_matrix(uc):
    """401 with no identity; 403 without the permission; served (anything but 401/403) with it."""
    assert call(uc, {}) == 401
    for role, perms in ROLES.items():
        got = call(uc, person(f"{role}@x", perms))
        if uc.permission in perms:
            assert got not in (401, 403), (role, uc.name, got)
        else:
            assert got == 403, (role, uc.name, got)
    assert call(uc, NOBODY) == 403


@pytest.mark.parametrize("grant", ["*:*:*", "ar:*:*", "ar:invoice:*", "ar:*:read"])
def test_a_wildcard_grant_runs_every_person_use_case_it_covers_and_never_a_machine_one(grant):
    who = person("wild@x", [grant])
    for uc in USE_CASES:
        got = call(uc, who)
        if uc.permission in MACHINE:
            assert got == 403, (grant, uc.name)
        elif grant in ("*:*:*", "ar:*:*"):
            assert got not in (401, 403), (grant, uc.name, got)


@pytest.mark.parametrize("grant", ["zz:*:*", "ar", "ar:*", "a*:invoice:read", "ar:invoice:*:*", ""])
def test_malformed_or_foreign_grants_run_nothing(grant):
    who = person("bad@x", [grant])
    assert all(call(uc, who) == 403 for uc in USE_CASES)


def test_a_person_holding_every_person_permission_still_cannot_run_a_machine_use_case():
    everyone = person("boss@x", sorted({p for perms in ROLES.values() for p in perms} - set(MACHINE)))
    assert [uc.name for uc in USE_CASES if uc.permission in MACHINE and call(uc, everyone) != 403] == []


def test_the_controls_are_never_held_by_whoever_they_control():
    """A biller or approver holding ar:settings:controls is a configuration error: the default roles keep it with the owner alone."""
    holders = [r for r, perms in ROLES.items() if "ar:settings:controls" in perms]
    assert holders == ["owner"]
    controlled = {"ar:invoice:write", "ar:invoice:issue", "ar:invoice:send", "ar:invoice:void", "ar:credit_note:write", "ar:credit_note:approve",
                  "ar:writeoff:write", "ar:writeoff:approve", "ar:settings:write", "ar:customer:write"}
    assert not controlled & set(ROLES["owner"])
    assert not {"ar:settings:controls", "ar:migration:sign"} & (set(ROLES["biller"]) | set(ROLES["approver"]))


def test_settings_write_cannot_change_a_control_and_controls_cannot_change_a_plain_setting():
    s, b = api("POST", "/settings", APPROVER, {"key": "void_policy", "value": "never"})
    assert s == 403 and "ar:settings:controls" in b["detail"]
    s, b = api("POST", "/controls", OWNER, {"key": "billing_day", "value": "month_start"})
    assert s == 403 and "not a control" in b["detail"]
    assert api("POST", "/controls", OWNER, {"key": "nope", "value": 1})[0] == 422
    assert api("POST", "/controls", OWNER, {"key": "void_policy", "value": "sometimes"})[0] == 422
    assert api("POST", "/settings", APPROVER, {"key": "billing_day", "value": "month_start"})[0] == 200
    got = must(api("GET", "/settings", BILLER))["settings"]
    assert "void_policy" not in got and got["billing_day"]["control"] is False and got["billing_day"]["value"] == "month_start"
    owner = must(api("GET", "/settings", OWNER))["settings"]
    assert owner["void_policy"]["control"] is True and owner["void_policy"]["value"] == "never_sent"


def test_sales_reads_status_and_balance_but_never_lines_and_cannot_issue_send_or_credit(world):
    inv = issued(world)
    assert api("GET", f"/customers/{BP}/status", SALES)[0] == 200
    for method, path in (("GET", f"/invoices/{inv['invoice_id']}"), ("GET", "/invoices"), ("POST", f"/invoices/{inv['invoice_id']}/issue"),
                         ("POST", f"/invoices/{inv['invoice_id']}/send"), ("POST", "/credit-notes"), ("POST", "/writeoffs"),
                         ("GET", f"/invoices/{inv['invoice_id']}/readiness"), ("GET", "/overdue")):
        assert api(method, path, SALES, {"reason": "r", "invoice_id": "x", "amount": "1.00"} if method == "POST" else None)[0] == 403, path
    narrow = must(api("GET", f"/customers/{BP}/status", SALES))
    assert not {"lines", "invoices", "total", "subtotal", "margin"} & set(narrow)


def test_a_draft_is_invisible_to_a_reader_who_does_not_author_invoices(world):
    billing()
    d = hand_draft()
    assert api("GET", f"/invoices/{d['invoice_id']}", COLLECTOR)[0] == 404                 # not 403: a 403 would confirm it exists
    assert api("GET", f"/invoices/{d['invoice_id']}", BILLER)[0] == 200
    assert api("GET", "/invoices", COLLECTOR, {"lifecycle": "draft"})[1]["invoices"] == []
    cn = must(api("POST", "/credit-notes", BILLER, {"bp_id": BP, "currency": "GBP", "reason": "goodwill", "lines": [line()]}))
    assert api("GET", f"/credit-notes/{cn['credit_id']}", COLLECTOR)[0] == 404


def test_a_biller_cannot_approve_and_an_author_cannot_approve_their_own_credit_where_a_second_person_is_asked(world):
    inv = issued(world)
    assert api("POST", "/credit-notes/x/approve", BILLER, {})[0] == 403
    must(control("second_approver", True))
    both = person("both@x", ["ar:credit_note:write", "ar:credit_note:approve", "ar:invoice:read", "ar:writeoff:write", "ar:writeoff:approve"])
    cn = must(api("POST", "/credit-notes", both, {"invoice_id": inv["invoice_id"], "reason": "r", "lines": [line("c", "1", "10.00")]}))
    s, b = api("POST", f"/credit-notes/{cn['credit_id']}/approve", both, {})
    assert s == 409 and "second_person" in b["detail"] and "ask another approver" in b["detail"]
    world.payments.credit_faces[cn["credit_id"]] = Decimal("12.00")
    assert must(api("POST", f"/credit-notes/{cn['credit_id']}/approve", APPROVER, {}))["status"] == "approved"
    w = must(api("POST", "/writeoffs", both, {"invoice_id": inv["invoice_id"], "amount": "1.00", "reason": "r",
             "received_on": "2026-11-20", "currency": "GBP", "discount_amount": "1.00", "net_amount": "1.00", "valid_from": "2026-06-30",
             "reference_rate_pct": "4.25", "rate_pct": "5", "period_id": "2026-10", "bucket": "0-30"}))
    s, b = api("POST", f"/writeoffs/{w['writeoff_id']}/approve", both, {"confirm_above_threshold": True})
    assert s == 409 and "second_person" in b["detail"]
    assert must(api("POST", f"/writeoffs/{w['writeoff_id']}/approve", APPROVER, {"confirm_above_threshold": True}))["status"] == "approved"


def test_only_the_migration_signature_signs_and_load_cannot_issue_post_or_send():
    loader = person("loader@x", ["ar:migration:load", "ar:migration:reconcile"])
    for method, path in (("POST", "/invoices/x/issue"), ("POST", "/documents/x/post"), ("POST", "/invoices/x/send"),
                         ("POST", "/migration/runs/x/sign"), ("POST", "/invoices/x/void")):
        assert api(method, path, loader, {"reason": "r"})[0] == 403, path
    for who in (BILLER, APPROVER, COLLECTOR, SYSTEM):
        assert api("POST", "/migration/runs/x/sign", who, {})[0] == 403
    assert api("POST", "/migration/runs/x/sign", OWNER, {})[0] == 404


def test_a_row_the_caller_may_not_see_is_a_404_never_a_403(world):
    assert api("GET", "/invoices/nope", BILLER)[0] == 404 and api("GET", "/credit-notes/nope", BILLER)[0] == 404
    assert api("GET", "/schedules/nope", BILLER)[0] == 404 and api("GET", "/migration/runs/nope", OWNER)[0] == 404


def test_the_mcp_door_lists_a_tool_per_use_case_and_refuses_what_the_route_refuses(world):
    names = tools(BILLER)
    assert {"draft_invoice", "issue_invoice", "invoice_readiness", "raise_draft", "approve_credit", "sign_reconciliation"} <= set(names)
    assert not [n for n in names if n.startswith("ar_")] and not [n for n in names if re.match(r"(update|delete|insert)_", n)]     # use cases, never table CRUD
    status, why = tool(SALES, "issue_invoice", {"invoice_id": "x"})
    assert status == "refused"
    status, body = tool(SALES, "customer_status", {"bp_id": str(BP)})
    assert status == "ok" and body["bp_id"] == str(BP)