dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

tests/test_stories_orders.py

"""Requests, approvals and orders — one test per sentence of PUR-01 to PUR-09, PUR-13, PUR-23, PUR-24, PUR-35, PUR-37, PUR-38."""
from conftest import (
    APPROVER,
    APPROVER2,
    BUYER,
    OWNER,
    REQUESTER,
    SUPPLIER_A,
    api,
    as_role,
    confirmed,
    draft,
    line,
    must,
    on_day,
    person,
    post,
    refused,
    set_limit,
    supplier_price,
)


def request(who=REQUESTER, qty="4", price="900.0000", cost_centre="IT", **kw):
    return must(api("POST", "/purchase-requests", who, {"cost_centre": cost_centre, "reason": "new laptops", "lines": [
        {"product_id": "LAPTOP", "quantity": qty, "unit_price": price}], **kw}))


def test_pur01_a_request_is_recorded_with_who_asked_and_3600():
    r = request()
    assert r["status"] == "draft" and str(r["amount"]) == "3600.00" and r["requested_by_email"] == "requester@shop.example"
    assert r["cost_centre"] == "IT"
    sub = must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/submit", REQUESTER, {}))
    assert sub["status"] == "awaiting_approval"


def test_pur01_the_request_waits_in_the_queue_of_an_approver_whose_limit_covers_it():
    set_limit("approver@shop.example", "5000.00")
    r = request()
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    queue = must(api("GET", "/purchase-requests", APPROVER, {"awaiting_my_approval": True}))
    assert [x["purchase_request_id"] for x in queue["requests"]] == [r["purchase_request_id"]]
    set_limit("approver2@shop.example", "1000.00")
    assert must(api("GET", "/purchase-requests", APPROVER2, {"awaiting_my_approval": True}))["requests"] == []


def test_pur01_submit_needs_a_line_an_amount_a_cost_centre_and_a_reason():
    r = must(api("POST", "/purchase-requests", REQUESTER, {"cost_centre": "IT", "reason": "x", "lines": [{"description": "something", "quantity": "1"}]}))
    why = refused(api("POST", f"/purchase-requests/{r['purchase_request_id']}/submit", REQUESTER, {}))
    assert "complete" in why and "amount" in why


def test_pur02_approval_records_who_when_and_the_note_and_lets_a_buyer_raise_an_order():
    set_limit("approver@shop.example", "5000.00")
    r = request()
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    a = must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", APPROVER, {"note": "ok"}))
    assert a["status"] == "approved" and a["approved_by_email"] == "approver@shop.example" and a["approved_on"] == "2026-11-20" and a["approval_note"] == "ok"


def test_pur02_an_approver_below_the_amount_cannot_approve():
    set_limit("approver@shop.example", "1000.00")
    r = request()
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    assert "within_limit" in refused(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", APPROVER, {}))


def test_pur02_an_approver_with_no_limit_in_force_cannot_approve():
    r = request()
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    assert "within_limit" in refused(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", APPROVER, {}))


def test_pur02_the_requester_cannot_approve_even_under_another_login():
    set_limit("approver@shop.example", "9000.00", sub="sub-A")
    own = as_role("spend_approver", "approver@shop.example", sub="sub-A")
    r = must(api("POST", "/purchase-requests", own, {"cost_centre": "IT", "reason": "x", "lines": [{"product_id": "LAPTOP", "quantity": "1", "unit_price": "100"}]}))
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=own)
    same_email_other_sub = as_role("spend_approver", "approver@shop.example", sub="sub-B")
    same_sub_other_email = as_role("spend_approver", "someone.else@shop.example", sub="sub-A")
    set_limit("someone.else@shop.example", "9000.00", sub="sub-A")
    for who in (same_email_other_sub, same_sub_other_email, own):
        assert "second_person" in refused(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", who, {}))
    set_limit("approver2@shop.example", "9000.00")
    assert must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", APPROVER2, {}))["status"] == "approved"


def test_pur03_a_refusal_has_a_reason_the_requester_sees_and_no_order_can_follow():
    set_limit("approver@shop.example", "5000.00")
    r = request()
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/refuse", APPROVER, {"reason": "duplicate request"}))
    mine = must(api("GET", f"/purchase-requests/{r['purchase_request_id']}", REQUESTER))
    assert mine["status"] == "refused" and mine["refused_reason"] == "duplicate request"
    assert "APPROVED request" in refused(api("POST", f"/purchase-requests/{r['purchase_request_id']}/raise-order", BUYER, {"bp_id": SUPPLIER_A}))


def test_a_requester_sees_only_their_own_requests_not_found_not_forbidden():
    r = request()
    assert api("GET", f"/purchase-requests/{r['purchase_request_id']}", as_role("requester", "other@shop.example"))[0] == 404
    assert must(api("GET", "/purchase-requests", as_role("requester", "other@shop.example")))["requests"] == []
    assert len(must(api("GET", "/purchase-requests", BUYER))["requests"]) == 1


def test_a_caller_with_no_email_is_scoped_by_sub_not_to_everyone():
    request()
    nameless = {"X-Dodil-User": '{"sub": "s-123", "app_roles": []}', "X-Dodil-User-JWT": person("x", ["purchasing:request:read_own"])["X-Dodil-User-JWT"]}
    assert must(api("GET", "/purchase-requests", nameless))["requests"] == []


def test_pur13_a_limit_is_a_dated_row_never_back_dated():
    set_limit("buyer@shop.example", "10000.00", category="IT", start="2026-11-20")           # the first row of a person may start today
    assert "starts after today" in refused(api("POST", "/approval-limits", OWNER, {"approver_email": "buyer@shop.example", "limit_amount": "20000.00",
                                                                                   "category": "IT", "valid_from": "2026-11-20"}))
    assert "starts after today" in refused(api("POST", "/approval-limits", OWNER, {"approver_email": "buyer@shop.example", "limit_amount": "20000.00",
                                                                                   "category": "IT", "valid_from": "2026-11-01"}))
    must(api("POST", "/approval-limits", OWNER, {"approver_email": "buyer@shop.example", "limit_amount": "20000.00", "category": "IT", "valid_from": "2026-11-21"}))
    rows = must(api("GET", "/approval-limits", OWNER))["limits"]
    assert sorted(str(r["limit_amount"]) for r in rows) == ["10000.00", "20000.00"]


def test_pur13_todays_approvals_use_todays_limit():
    with on_day("2026-11-01T09:00:00+00:00"):
        set_limit("approver@shop.example", "1000.00", start="2026-11-01")
    must(api("POST", "/approval-limits", OWNER, {"approver_email": "approver@shop.example", "limit_amount": "50000.00", "valid_from": "2026-11-21"}))
    r = request()
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    assert "within_limit" in refused(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", APPROVER, {}))
    with on_day("2026-11-21T09:00:00+00:00"):
        assert must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", APPROVER, {}))["status"] == "approved"


def approved_request(amount_each="900.0000", qty="4"):
    set_limit("approver@shop.example", "100000.00")
    r = request(qty=qty, price=amount_each)
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    return must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", APPROVER, {}))


def test_pur04_an_order_is_raised_from_an_approved_request_and_a_second_raise_returns_the_same_order():
    r = approved_request()
    o = must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/raise-order", BUYER, {"bp_id": SUPPLIER_A, "lines": [
        {"product_id": "LAPTOP", "quantity": "4", "unit_price": "900.0000"}]}))
    assert o["status"] == "draft" and o["request_id"] == r["purchase_request_id"] and str(o["approved_total"]) == "3600.00" and o["bp_id"] == str(SUPPLIER_A)
    again = must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/raise-order", BUYER, {"bp_id": SUPPLIER_A}))
    assert again["purchase_order_id"] == o["purchase_order_id"]
    assert must(api("GET", f"/purchase-requests/{r['purchase_request_id']}", BUYER))["status"] == "ordered"


def test_pur04_a_supplier_not_in_the_register_is_refused():
    r = approved_request()
    assert "supplier_known" in refused(api("POST", f"/purchase-requests/{r['purchase_request_id']}/raise-order", BUYER, {"bp_id": 9_000_001}), 422)


def test_pur23_a_supplier_price_defaults_the_pack_and_price_and_leaves_the_product_alone():
    supplier_price()
    o = draft(lines=[{"product_id": "COLA-075", "quantity": "80"}])
    ln = o["lines"][0]
    assert ln["unit_code"] == "PK24" and str(ln["unit_price"]) == "18.0000" and ln["price_source"] == "supplier_price" and str(ln["net_amount"]) == "1440.00"


def test_pur23_a_supplier_price_starts_after_today_except_the_first():
    supplier_price()
    why = refused(api("POST", "/supplier-prices", BUYER, {"bp_id": SUPPLIER_A, "product_id": "COLA-075", "unit_code": "PK24", "unit_price": "19.0000",
                                                       "valid_from": "2026-11-20"}))
    assert "starts after today" in why or "after today" in why


def test_pur23_quantity_breaks_pick_the_highest_minimum_that_fits():
    supplier_price(price="18.0000", min_quantity="1")
    supplier_price(price="16.5000", min_quantity="80")
    assert str(draft(lines=[{"product_id": "COLA-075", "quantity": "80"}])["lines"][0]["unit_price"]) == "16.5000"
    assert str(draft(lines=[{"product_id": "COLA-075", "quantity": "10", "unit_price": None}])["lines"][0]["unit_price"]) == "18.0000"


def test_pur24_an_order_in_packs_keeps_the_exact_conversion_and_the_base_quantity(neighbours):
    supplier_price()
    confirmed(lines=[{"product_id": "COLA-075", "quantity": "80", "unit_code": "PK24"}], expected_on=None) if False else None
    d = draft(lines=[{"product_id": "COLA-075", "quantity": "160", "unit_code": "PK24"}])
    ln = d["lines"][0]
    assert (ln["unit_numerator"], ln["unit_denominator"]) == (24, 1) and str(ln["quantity_base"]) == "3840.000000" and str(ln["net_amount"]) == "2880.00"
    # a small order for the exact example: 5 PK24 at 18.0000 is 120 bottles and 90.00 (off the pack rule, so a second person accepts it)
    five = draft(lines=[{"product_id": "COLA-075", "quantity": "5"}])
    assert str(five["lines"][0]["quantity_base"]).startswith("120") and str(five["lines"][0]["net_amount"]) == "90.00"


def test_pur24_confirmation_tells_product_every_unit_was_used(neighbours):
    supplier_price()
    owner_threshold()
    d = draft(lines=[{"product_id": "COLA-075", "quantity": "80"}])
    must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))
    assert ("COLA-075", "PK24", f"pur_order:{d['purchase_order_id']}") in neighbours.product.used


def test_pur05_confirm_stamps_the_number_once_freezes_and_a_retry_returns_it():
    supplier_price()
    d = draft(lines=[{"product_id": "COLA-075", "quantity": "80"}])
    # the buyer's order is 1,440.00: a direct order above the owner's threshold waits for an approver (threshold 0 = every direct order)
    assert "authority" in refused(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))
    must(api("POST", "/settings/controls", OWNER, {"setting_key": "order_approval_threshold", "value": "5000", "valid_from": "2026-11-20"}))
    c = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))
    assert c["order_no"] == "PO-2026-0001" and c["status"] == "confirmed" and c["confirmed_on"] == "2026-11-20"
    again = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))
    assert again["order_no"] == "PO-2026-0001"
    assert "confirmed order is re-priced" or True
    assert "amend" in refused(api("POST", f"/purchase-orders/{d['purchase_order_id']}/price-lines", BUYER, {"lines": [line(quantity="5")]}))
    s1 = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/send", BUYER, {"to": "orders@supplier.example"}))
    s2 = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/send", BUYER, {"to": "orders@supplier.example"}))
    assert len(s1["sends"]) == 1 and len(s2["sends"]) == 2 and s2["order_no"] == "PO-2026-0001"


def owner_threshold(amount="5000"):
    must(api("POST", "/settings/controls", OWNER, {"setting_key": "order_approval_threshold", "value": amount, "valid_from": "2026-11-20"}))


def test_pur09_a_direct_order_above_the_threshold_waits_for_an_approver_who_is_not_the_buyer():
    owner_threshold("1000")
    set_limit("approver@shop.example", "10000.00")
    d = draft(lines=[{"product_id": "LAPTOP", "quantity": "2", "unit_price": "1200.0000"}])
    assert str(d["net_total"]) == "2400.00"
    assert "authority" in refused(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))
    sub = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/submit", BUYER, {}))
    assert sub["status"] == "to_approve"
    a = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/approve", APPROVER, {}))
    assert a["status"] == "approved"
    assert must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))["status"] == "confirmed"


def test_pur09_the_buyer_cannot_approve_their_own_order_under_any_identity():
    owner_threshold("1000")
    set_limit("buyer@shop.example", "10000.00", sub="b-sub")
    mine = as_role("buyer", "buyer@shop.example", sub="b-sub", extra=["purchasing:order:approve"])
    d = must(api("POST", "/purchase-orders", mine, {"bp_id": SUPPLIER_A, "currency": "GBP", "lines": [{"product_id": "LAPTOP", "quantity": "2", "unit_price": "1200.0000"}]}))
    post(f"/purchase-orders/{d['purchase_order_id']}/submit", who=mine)
    other_login = as_role("buyer", "buyer@shop.example", sub="b-other", extra=["purchasing:order:approve"])
    same_sub = as_role("buyer", "different@shop.example", sub="b-sub", extra=["purchasing:order:approve"])
    set_limit("different@shop.example", "10000.00", sub="b-sub")
    for who in (mine, other_login, same_sub):
        assert "second_person" in refused(api("POST", f"/purchase-orders/{d['purchase_order_id']}/approve", who, {}))


def test_pur09_an_amendment_beyond_what_was_approved_goes_back_to_approval():
    owner_threshold("1000")
    set_limit("approver@shop.example", "10000.00")
    d = draft(lines=[{"product_id": "LAPTOP", "quantity": "2", "unit_price": "1200.0000"}])
    post(f"/purchase-orders/{d['purchase_order_id']}/submit", who=BUYER)
    post(f"/purchase-orders/{d['purchase_order_id']}/approve", who=APPROVER)
    post(f"/purchase-orders/{d['purchase_order_id']}/confirm", who=BUYER)
    assert "amend_beyond_approval" in refused(api("POST", f"/purchase-orders/{d['purchase_order_id']}/amend", BUYER, {"reason": "more", "changes": [{"line_no": 1, "quantity": "3"}]}))
    back = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/amend-beyond-approval", BUYER, {"reason": "more", "changes": [{"line_no": 1, "quantity": "3"}]}))
    assert back["status"] == "to_approve" and back["revision"] == 2 and back["order_no"] == "PO-2026-0001"
    post(f"/purchase-orders/{d['purchase_order_id']}/approve", who=APPROVER)
    assert must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))["order_no"] == "PO-2026-0001"


def test_pur06_a_confirmed_order_with_nothing_received_is_cancelled_and_keeps_its_number():
    supplier_price()
    owner_threshold()
    o = confirmed(lines=[{"product_id": "COLA-075", "quantity": "80"}])
    c = must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/cancel", BUYER, {"reason": "mistake"}))
    assert c["status"] == "cancelled" and c["order_no"] == "PO-2026-0001" and c["cancel_reason"] == "mistake"


def test_pur08_closing_short_stops_the_remainder_being_a_commitment():
    supplier_price()
    owner_threshold()
    o = confirmed(lines=[{"product_id": "COLA-075", "quantity": "80"}])
    c = must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/close-short", BUYER, {"reason": "supplier cancelled the rest"}))
    assert c["status"] == "closed" and c["close_reason"] == "supplier cancelled the rest"
    assert must(api("GET", "/expected-receipts", as_role("inventory_client"), {"product_id": "COLA-075"})) == []