dodil/erp-brick-sales-order

DODIL ERP brick: sales order

tests/test_access.py

"""Access rules (module.yaml tests.access) and the generated probes: who may run each use case, and which rows they see."""
import pytest
from conftest import (
    ADMIN,
    ADMIN2,
    ALICE,
    FULFILLER,
    NOBODY,
    OWNER,
    REP,
    REP2,
    SYSTEM,
    api,
    confirmed,
    hand,
    must,
    person,
    refused,
    tool,
    tools,
)

from brick_sales_order.application import use_cases

W = [{"product_id": "WIDGET", "quantity": "12", "unit_code": "C62"}]


def test_a_rep_sees_and_changes_only_their_own_orders_another_reps_order_is_a_404():
    o = hand(REP)
    sid = o["sales_order_id"]
    assert api("GET", f"/sales-orders/{sid}", REP2)[0] == 404
    for method, path, body in [("POST", "confirm", {}), ("POST", "cancel-draft", {}), ("POST", "recalculate-tax", {}),
                               ("POST", "customer-po", {"customer_po_ref": "X"}), ("POST", "lines", {"line": {"product_id": "FEE", "quantity": "1"}}),
                               ("GET", "readiness", None)]:
        assert api(method, f"/sales-orders/{sid}/{path}", REP2, body)[0] == 404, path
    assert api("GET", f"/sales-orders/{sid}", ADMIN)[0] == 200                                    # read_all sees every order
    assert sid not in {r["sales_order_id"] for r in must(api("GET", "/sales-orders", REP2))["orders"]}
    assert tool(REP2, "get_sales_order", {"sales_order_id": sid})[0] == "refused"                # an agent sees what its user sees


def test_a_fulfiller_records_deliveries_and_never_sees_a_price():
    o = confirmed(lines=W)
    assert must(api("POST", f"/sales-orders/{o['sales_order_id']}/deliveries", FULFILLER, {"line_no": 1, "quantity": "1", "delivery_ref": "D1"}))["deliveries"]
    assert "unit_price" not in must(api("GET", f"/sales-orders/{o['sales_order_id']}", FULFILLER))["lines"][0]
    assert "unit_price" not in tool(FULFILLER, "get_sales_order", {"sales_order_id": o["sales_order_id"]})[1]["lines"][0]
    assert api("POST", f"/sales-orders/{o['sales_order_id']}/confirm", FULFILLER, {})[0] == 403


def test_only_a_holder_of_cancel_cancels_and_only_approve_overrides_a_credit_hold(neighbours):
    o = confirmed(lines=W)
    assert api("POST", f"/sales-orders/{o['sales_order_id']}/cancel", REP, {"reason": "x"})[0] == 403
    assert api("POST", f"/sales-orders/{o['sales_order_id']}/cancel", OWNER, {"reason": "x"})[0] == 403
    assert api("POST", f"/sales-orders/{o['sales_order_id']}/cancel", ADMIN, {"reason": "x"})[0] == 200
    d = hand()
    assert api("POST", f"/sales-orders/{d['sales_order_id']}/override-credit-hold", REP, {"reason": "x"})[0] == 403


def test_only_the_controls_permission_changes_a_control_and_write_is_refused_on_them():
    for key, value in [("confirm_on_acceptance", True), ("amendment_needs_approval", False), ("lock_done_orders", False),
                       ("business_timezone", "Europe/Paris"), ("ai.ai_order_from_po.mode", "suggest")]:
        assert api("POST", f"/settings/{key}", ADMIN, {"value": value, "valid_from": "2026-11-21"})[0] == 403, key
        assert api("POST", f"/controls/{key}", ADMIN, {"value": value, "valid_from": "2026-11-21"})[0] == 403, key
        assert api("POST", f"/controls/{key}", REP, {"value": value, "valid_from": "2026-11-21"})[0] == 403, key
        assert api("POST", f"/controls/{key}", OWNER, {"value": value, "valid_from": "2026-11-21"})[0] == 200, key
    assert "not a control" in refused(api("POST", "/controls/order_number_format", OWNER, {"value": "X", "valid_from": "2026-11-21"}), 404)


def test_a_wildcard_grant_does_not_reach_the_machine_permissions():
    star = person("star@shop.example", ["so:*:*"])
    every = person("every@shop.example", ["*:*:*"])
    other = person("other@shop.example", ["pj:*:*"])
    quote = {"quote_id": "Q", "quote_version": 1, "bp_id": 7311, "currency": "GBP", "lines": [{"line_no": 1, "product_id": "FEE", "quantity": "1", "list_price": "1", "unit_price": "1", "priced_from": {"price_list_id": "LIST-GBP", "rule_seq": 1}}]}
    for who in (star, every):
        assert api("POST", "/sales-orders/from-quote", who, quote)[0] == 403
        assert api("POST", "/sales-orders/ai-order-from-po", who, BODY["ai_order_from_po"])[0] == 403
        assert api("POST", "/sales-orders/x/finish", who, {})[0] == 403
        assert api("POST", "/sales-orders/x/ai-check-po", who, BODY["ai_check_po"])[0] == 403
        assert api("POST", "/number-series/SO-2026", who, BODY["set_number_series_mark"])[0] == 403
        assert api("POST", "/migration/odoo/import", who, BODY["import_odoo_orders"])[0] == 403
        assert api("GET", "/migration/odoo/reconcile", who)[0] == 403
        assert api("POST", "/sales-orders/x/return-inspections", who, {"line_no": 1, "inspection_ref": "i", "received_quantity": "1", "accepted_quantity": "1",
                                                                        "rejected_quantity": "0", "decision": "restock"})[0] == 403
        assert api("GET", "/sales-orders", who)[0] == 200                                         # but every person command passes
        assert api("POST", "/sales-orders", who, {"bp_id": 7311, "currency": "GBP", "lines": [{"product_id": "FEE", "quantity": "1"}]})[0] == 200
    assert api("GET", "/sales-orders", other)[0] == 403                                           # a wildcard of another module reaches nothing
    assert api("POST", "/sales-orders/from-quote", SYSTEM, quote)[0] == 200                       # the exact grant does


def test_an_amendment_is_never_approved_by_the_rep_who_made_it():
    o = confirmed(who=ALICE, lines=W)
    must(api("POST", f"/sales-orders/{o['sales_order_id']}/amend", ALICE, {"reason": "r", "changes": [{"line_no": 1, "quantity": "13"}]}))
    assert "approver_not_amender" in refused(api("POST", f"/sales-orders/{o['sales_order_id']}/revisions/2/approve", ALICE, {}))
    assert api("POST", f"/sales-orders/{o['sales_order_id']}/revisions/2/approve", ADMIN2, {})[0] == 200


def _concrete(uc):
    path = uc.path
    for p in uc.path_params:
        path = path.replace("{" + p + "}", "1" if p in ("revision", "line_no", "suggestion_no", "settlement_no", "milestone_no") else "nope")
    return path.replace("{step}", "ai_check_po")


LINE = {"product_id": "FEE", "quantity": "1"}
BODY = {
    "draft_by_hand": {"bp_id": 7311, "currency": "GBP", "lines": [LINE]}, "add_order_line": {"line": LINE},
    "record_customer_po": {"customer_po_ref": "PO"}, "override_credit_hold": {"reason": "r"},
    "order_from_quote": {"quote_id": "Q", "quote_version": 1, "bp_id": 7311, "currency": "GBP",
                         "lines": [{"line_no": 1, "product_id": "FEE", "quantity": "1", "list_price": "1", "unit_price": "1", "priced_from": {"price_list_id": "LIST-GBP", "rule_seq": 1}}]},
    "send_confirmation": {"address": "a@b.c"}, "record_delivery": {"line_no": 1, "quantity": "1", "delivery_ref": "D"},
    "amend_order": {"reason": "r", "changes": [{"line_no": 1, "quantity": "2"}]}, "cancel_order": {"reason": "r"},
    "ai_order_from_po": {"bp_id": 7311, "currency": "GBP", "customer_po_ref": "P", "owner": "o", "model": "m", "inputs_ref": "i", "suggestion": {}},
    "ai_check_po": {"model": "m", "inputs_ref": "i", "differences": []}, "accept_order_suggestion": {"decisions": []},
    "set_so_setting": {"value": 1, "valid_from": "2026-12-01"}, "set_so_control": {"value": True, "valid_from": "2026-12-01"},
    "set_customer_order_policy": {"bp_id": 7311, "valid_from": "2026-12-01"}, "set_number_series_mark": {"high_water_mark": 3},
    "import_odoo_orders": {"cut_off": "2026-11-20", "migration_owner": "boss@shop.example"},
    "set_order_type": {"order_type_code": "RUSH", "valid_from": "2026-12-01"},
    "set_item_category_rule": {"order_type_code": "RUSH", "product_type_code": "TRADING", "usage": "normal", "item_category_code": "STD_STOCK",
                               "valid_from": "2026-12-01"},
    "add_condition_record": {"condition_type_code": "DISC_PCT", "key_fields": "category", "valid_from": "2026-12-01", "rate": "1", "category_code": "x"},
    "set_manual_condition": {"condition_type_code": "DISC_PCT", "rate": "1"}, "set_header_condition": {"condition_type_code": "HDR_DISC", "rate": "1"},
    "change_order_type": {"order_type_code": "RUSH"},
    "open_rebate_agreement": {"valid_from": "2027-01-01", "valid_to": "2027-12-31", "basis": "all", "scale": [{"from_value": "0", "rate_pct": "1"}],
                              "accrual_rate_pct": "1", "currency": "GBP", "bp_id": 7311},
    "set_customer_partner": {"sold_to_bp_id": 7311, "function": "payer", "partner_bp_id": 7400, "valid_from": "2026-12-01"},
    "add_incompletion_rule": {"order_type_code": "STANDARD", "field": "delivery_terms", "valid_from": "2026-12-01", "blocks": "delivery",
                              "condition": "always", "reason_text": "r"},
    "add_block_reason": {"block_reason_code": "X", "kind": "delivery", "name": "x"},
    "set_order_block": {"kind": "delivery", "reason_code": "X", "note": "n"}, "release_order_block": {"kind": "delivery", "note": "n"},
    "request_partial_settlement": {"reason": "r"}, "request_final_settlement": {"reason": "r"},
    "change_rebate_scale": {"scale": [{"from_value": "0", "rate_pct": "1"}]},
    "record_return_inspection": {"line_no": 1, "inspection_ref": "inv:RET-1:1", "received_quantity": "1", "accepted_quantity": "1", "rejected_quantity": "0",
                                 "decision": "restock"},
    "cancel_return": {"reason": "r"}, "set_return_reason": {"return_reason_code": "X", "name": "x"},
}
QUERY = {"get_demand": {"product_id": "FEE", "warehouse_id": "WH1", "from": "2026-11-01", "to": "2026-11-30"}}      # a GET with required query parameters
USE_CASES = [(uc.name, uc.method, _concrete(uc), uc.permission) for uc in use_cases.SO.use_cases if uc.permission]


@pytest.mark.parametrize("name,method,path,permission", USE_CASES)
def test_the_permission_matrix_401_without_identity_403_without_the_permission(name, method, path, permission):
    body_in = BODY.get(name, {}) if method != "GET" else QUERY.get(name)
    assert api(method, path, {}, body_in)[0] == 401, name
    status, body = api(method, path, NOBODY, body_in)
    assert status == 403 and permission in str(body), (name, status, body)


def test_every_use_case_is_a_route_and_a_tool_except_the_machine_only_ones():
    status, spec = api("GET", "/openapi.json", ALICE)
    assert status == 200
    listed = set(tools(ALICE))
    for uc in use_cases.SO.use_cases:
        assert uc.method.lower() in spec["paths"].get(uc.path.replace("{", "{"), {}), uc.name
        assert (uc.name in listed) == uc.tool, uc.name
    assert {"order_from_quote", "finish_order", "ai_order_from_po", "ai_check_po", "set_number_series_mark"}.isdisjoint(listed)


def test_a_write_without_the_permission_writes_nothing():
    assert api("POST", "/sales-orders", NOBODY, {"bp_id": 7311, "currency": "GBP", "lines": W, "sales_order_id": "NOPE"})[0] == 403
    assert api("GET", "/sales-orders/NOPE", ALICE)[0] == 404


def test_an_extra_field_is_never_applied():
    o = hand()
    status, body = api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "lines": W, "owner": "mallory@x", "total": "0.01", "lifecycle": "confirmed"})
    assert status in (200, 422)
    if status == 200:
        assert body["owner"] == "sales_rep@shop.example" and body["lifecycle"] == "draft" and body["total"] != "0.01"
    assert o