dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

tests/test_stories_sourcing.py

"""Sourcing — the source list (PUR-44), the quota (PUR-45), requests for quotation (PUR-39, PUR-40)."""
from conftest import (
    BUYER,
    CLERK,
    REQUESTER,
    SUPPLIER_A,
    SUPPLIER_B,
    api,
    as_role,
    confirmed,
    must,
    on_day,
    post,
    supplier_price,
    world,
)


def test_pur44_a_blocked_or_unlisted_source_is_refused_at_pricing():
    world()
    with on_day("2026-01-01T09:00:00+00:00"):
        for bp, blocked, fixed in ((SUPPLIER_B, False, True), (7003, False, False), (7004, True, False)):
            must(api("POST", "/source-list", BUYER, {"product_id": "STEEL-BAR", "bp_id": bp, "valid_from": "2026-01-02" if bp != 7004 else "2026-12-01",
                                                      "fixed": fixed, "blocked": blocked, "required": bp == SUPPLIER_B}))
    for bp in (7004, 7005):
        s, body = api("POST", "/purchase-orders", BUYER, {"bp_id": bp, "currency": "GBP", "lines": [{"product_id": "STEEL-BAR", "quantity": "10", "unit_code": "C62", "unit_price": "3.2"}]})
        assert s == 409 and "source_allowed" in str(body)
    with on_day("2026-12-02T09:00:00+00:00"):
        s, body = api("POST", "/purchase-orders", BUYER, {"bp_id": 7004, "currency": "GBP", "lines": [{"product_id": "STEEL-BAR", "quantity": "10", "unit_code": "C62", "unit_price": "3.2"}]})
        assert s == 409 and "blocked from 2026-12-01" in str(body) and "7003" in str(body)


def test_pur44_a_product_without_a_required_list_may_be_bought_from_anyone_and_rows_start_after_today():
    world()
    must(api("POST", "/source-list", BUYER, {"product_id": "WIDGET", "bp_id": SUPPLIER_A, "valid_from": "2026-11-20"}))          # the first row may start today
    s, b = api("POST", "/source-list", BUYER, {"product_id": "WIDGET", "bp_id": SUPPLIER_B, "valid_from": "2026-11-20"})
    assert s == 409 and "starts after today" in str(b) or "after today" in str(b)
    s, o = api("POST", "/purchase-orders", BUYER, {"bp_id": 7009, "currency": "GBP", "lines": [{"product_id": "WIDGET", "quantity": "1", "unit_code": "C62", "unit_price": "2"}]})
    assert s == 200
    assert api("POST", "/source-list", CLERK, {"product_id": "WIDGET", "bp_id": 1, "valid_from": "2026-12-01"})[0] == 403


def _quota_world(setup=True):
    if setup:
        world()
    for bp in (SUPPLIER_A, SUPPLIER_B):
        supplier_price(bp=bp, product="LAPTOP", unit="C62", price="1.0000")
    must(api("POST", "/quotas", BUYER, {"product_id": "LAPTOP", "valid_from": "2026-11-20", "shares": [{"bp_id": SUPPLIER_A, "quota_pct": "70"}, {"bp_id": SUPPLIER_B, "quota_pct": "30"}]}))
    confirmed(bp=SUPPLIER_A, lines=[{"product_id": "LAPTOP", "quantity": "1400"}])
    confirmed(bp=SUPPLIER_B, lines=[{"product_id": "LAPTOP", "quantity": "400"}])


def test_pur45_the_lowest_rating_is_proposed_with_both_ratings_shown():
    _quota_world()
    p = must(api("GET", "/quota-proposal", BUYER, {"product_id": "LAPTOP", "quantity": "200"}))
    assert p["proposed"]["bp_id"] == str(SUPPLIER_B)
    by = {int(c["bp_id"]): c for c in p["candidates"]}
    assert str(by[SUPPLIER_A]["rating"]) in ("20", "20.0000") and str(by[SUPPLIER_B]["rating"]).startswith("13.3333")
    assert str(by[SUPPLIER_A]["allocated"]).startswith("1400") and by[SUPPLIER_B]["blocked"] is False


def test_pur45_allocated_is_derived_from_orders_not_a_counter_and_a_blocked_source_is_never_proposed():
    _quota_world()
    confirmed(bp=SUPPLIER_B, lines=[{"product_id": "LAPTOP", "quantity": "1000"}])          # B now 1400 / 30 = 46.7 -> A (20.0) is proposed
    assert must(api("GET", "/quota-proposal", BUYER, {"product_id": "LAPTOP", "quantity": "1"}))["proposed"]["bp_id"] == str(SUPPLIER_A)
    must(api("POST", "/source-list", BUYER, {"product_id": "LAPTOP", "bp_id": SUPPLIER_A, "valid_from": "2026-11-21", "blocked": True}))
    with on_day("2026-11-22T09:00:00+00:00"):
        assert must(api("GET", "/quota-proposal", BUYER, {"product_id": "LAPTOP", "quantity": "1"}))["proposed"]["bp_id"] == str(SUPPLIER_B)


def test_pur45_shares_that_do_not_sum_to_100_are_refused_at_entry_and_a_request_without_a_supplier_takes_the_proposal():
    world()
    s, b = api("POST", "/quotas", BUYER, {"product_id": "LAPTOP", "valid_from": "2026-11-20", "shares": [{"bp_id": SUPPLIER_A, "quota_pct": "70"}, {"bp_id": SUPPLIER_B, "quota_pct": "20"}]})
    assert s == 422 and "not 100" in str(b)
    _quota_world(setup=False)
    r = must(api("POST", "/purchase-requests", REQUESTER, {"cost_centre": "IT", "reason": "restock", "lines": [{"product_id": "LAPTOP", "quantity": "200", "unit_price": "1.0000"}]}))
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", as_role("spend_approver", "approver@shop.example"), {}))
    o = must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/raise-order", BUYER, {}))
    assert o["bp_id"] == str(SUPPLIER_B)


def _approved_request(qty="200"):
    r = must(api("POST", "/purchase-requests", REQUESTER, {"cost_centre": "IT", "reason": "chairs", "lines": [{"product_id": "LAPTOP", "quantity": qty, "unit_price": "150.0000"}]}))
    post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
    must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", as_role("spend_approver", "approver@shop.example"), {}))
    return r["purchase_request_id"]


def _rfq(deadline="2026-11-30"):
    world()
    rid = _approved_request()
    r = must(api("POST", "/rfqs", BUYER, {"request_id": rid, "deadline": deadline}))
    return r["rfq_id"], rid


def test_pur39_an_rfq_is_sent_to_three_suppliers_numbered_and_sending_twice_changes_nothing_else():
    rfq, _ = _rfq()
    inv = [{"bp_id": b, "to": f"s{b}@x.example"} for b in (7001, 7002, 7003)]
    r = must(api("POST", f"/rfqs/{rfq}/send", BUYER, {"invitations": inv}))
    assert r["rfq_no"] == "RFQ-2026-0001" and len(r["invitations"]) == 3 and r["lines"][0]["quantity"] and all(i["sends"] == 1 for i in r["invitations"])
    again = must(api("POST", f"/rfqs/{rfq}/send", BUYER, {"invitations": inv[:1]}))
    assert again["rfq_no"] == r["rfq_no"] and [i["sends"] for i in again["invitations"]] == [2, 1, 1] and again["status"] == "sent"


def test_pur39_a_deadline_today_or_earlier_and_a_blocked_supplier_are_refused():
    world()
    rid = _approved_request()
    assert api("POST", "/rfqs", BUYER, {"request_id": rid, "deadline": "2026-11-20"})[0] == 422
    must(api("POST", "/source-list", BUYER, {"product_id": "LAPTOP", "bp_id": 7003, "valid_from": "2026-11-20", "blocked": True}))
    rfq = must(api("POST", "/rfqs", BUYER, {"request_id": rid, "deadline": "2026-11-30"}))["rfq_id"]
    s, b = api("POST", f"/rfqs/{rfq}/send", BUYER, {"invitations": [{"bp_id": 7003, "to": "c@x.example"}]})
    assert s == 409 and "LAPTOP" in str(b) and "source_allowed" in str(b)
    assert api("POST", f"/rfqs/{rfq}/send", CLERK, {"invitations": []})[0] == 403


def _quoted():
    rfq, rid = _rfq()
    must(api("POST", f"/rfqs/{rfq}/send", BUYER, {"invitations": [{"bp_id": b, "to": f"s{b}@x.example"} for b in (7001, 7002, 7003)]}))
    must(api("POST", f"/rfqs/{rfq}/quotes", BUYER, {"bp_id": 7001, "lines": [{"line_no": 1, "unit_price": "140.0000", "delivery_days": 10}]}))
    must(api("POST", f"/rfqs/{rfq}/quotes", BUYER, {"bp_id": 7002, "lines": [{"line_no": 1, "unit_price": "132.0000", "delivery_days": 30}]}))
    return rfq, rid


def test_pur40_the_comparison_marks_the_cheapest_and_lists_who_did_not_quote():
    rfq, _ = _quoted()
    c = must(api("GET", f"/rfqs/{rfq}", BUYER))["comparison"]
    assert c["cheapest_bp_id"] == "7002" and c["no_quote"] == ["7003"]
    assert {int(s["bp_id"]): s["net"] for s in c["suppliers"]} == {7001: "28000.00", 7002: "26400.00"}
    with on_day("2026-12-01T09:00:00+00:00"):
        late = must(api("POST", f"/rfqs/{rfq}/quotes", BUYER, {"bp_id": 7003, "lines": [{"line_no": 1, "unit_price": "150.0000"}]}))
    assert next(s for s in late["comparison"]["suppliers"] if s["bp_id"] == "7003")["late"] is True


def test_pur40_awarding_a_dearer_quote_needs_a_reason_and_drafts_the_keyed_order_once():
    rfq, rid = _quoted()
    s, b = api("POST", f"/rfqs/{rfq}/award", BUYER, {"bp_id": 7001})
    assert s == 409 and "cheapest" in str(b)
    a = must(api("POST", f"/rfqs/{rfq}/award", BUYER, {"bp_id": 7001, "reason": "B cannot deliver before the office opens"}))
    o = a["order"]
    assert o["purchase_order_id"] == f"rfq:{rfq}:7001" and o["status"] == "draft" and str(o["lines"][0]["unit_price"]) == "140.0000" and o["lines"][0]["price_source"] == "quote"
    assert a["to_tell"] == ["7002", "7003"] and o["approved_total"] and str(o["total"]) == "28000.00"
    again = must(api("POST", f"/rfqs/{rfq}/award", BUYER, {"bp_id": 7001, "reason": "x"}))
    assert again["order"]["purchase_order_id"] == o["purchase_order_id"]
    assert api("POST", f"/rfqs/{rfq}/award", BUYER, {"bp_id": 7003})[0] == 409