dodil/erp-brick-sales-order

DODIL ERP brick: sales order

tests/test_audit_v6.py

"""Audit of sales-order v6 (d276341..2d0fe6f): each MUST is proven by a strict-xfail test that fails now and passes when fixed."""
from decimal import Decimal as D

from conftest import (
    BILLER,
    INVENTORY,
    REP,
    api,
    as_role,
    confirm,
    deliver,
    get,
    hand,
    must,
    on_day,
    refused,
)
from test_slice_b import warehouse_set
from test_slice_c import CONTROLLER, PRINTER, original, ret

COLA6 = {"product_id": "COLA-075", "quantity": "6", "unit_code": "C62"}


def test_MUST_2_a_confirm_refused_by_inventory_has_asked_purchasing_for_nothing(neighbours, inventory):
    N = neighbours
    warehouse_set("WH1")
    o = hand(REP, ship_to_bp_id=7313, lines=[*PRINTER, COLA6])
    inventory.down = True
    assert api("POST", f"/sales-orders/{o['sales_order_id']}/confirm", REP, {})[0] >= 400
    assert get(o["sales_order_id"])["lifecycle"] == "draft"
    must(api("POST", f"/sales-orders/{o['sales_order_id']}/cancel-draft", REP, {}))
    assert N.purchasing.requests == {} or N.purchasing.cancelled, "purchasing holds a request for a draft that was then cancelled"


def test_MUST_2_a_cancel_rolled_back_by_inventory_has_not_cancelled_purchasings_request(neighbours, inventory):
    N = neighbours
    warehouse_set("WH1")
    o = confirm(hand(REP, ship_to_bp_id=7313, lines=[*PRINTER, COLA6]))
    inventory.down = True
    assert api("POST", f"/sales-orders/{o['sales_order_id']}/cancel", as_role("order_admin"), {"reason": "withdrawn"})[0] >= 400
    assert get(o["sales_order_id"])["lifecycle"] == "confirmed"
    assert N.purchasing.cancelled == [], "the order is still confirmed but its drop-ship request is cancelled in purchasing"


def test_MUST_1_cancelling_after_a_draft_invoice_deducted_the_deposit_credits_the_whole_deposit(neighbours):
    from test_slice_c import bill, deposit
    N = neighbours
    o = confirm(hand(REP, lines=[{"product_id": "FEE", "quantity": "10", "unit_code": "C62"}]))      # 10,000.00, policy ordered
    sid = o["sales_order_id"]
    d1 = deposit(N, o, "5000.00")
    must(bill(o))                                                                                     # a draft that deducts the deposit
    assert N.invoicing.drafts[f"so:{sid}:2"]["deductions"]
    out = must(api("POST", f"/sales-orders/{sid}/cancel", as_role("order_admin"), {"reason": "the customer withdrew"}))
    assert out["lifecycle"] == "cancelled" and f"so:{sid}:2" in N.invoicing.cancelled
    credited = sum(D(x["net_amount"]) for c in N.invoicing.credits.values() if c["invoice_id"] == d1 for x in c["lines"])
    assert credited == D("5000.00"), f"the customer paid a 5,000.00 deposit and is credited {credited}"


def test_MUST_8_amending_a_drop_ship_line_reorders_from_purchasing_at_the_new_quantity(neighbours, inventory):
    N = neighbours
    o = confirm(hand(REP, ship_to_bp_id=7313, lines=PRINTER))
    sid = o["sales_order_id"]
    must(api("POST", f"/sales-orders/{sid}/amend", REP, {"reason": "they want five", "changes": [{"line_no": 1, "quantity": "5"}]}))
    must(api("POST", f"/sales-orders/{sid}/revisions/2/approve", as_role("order_admin"), {}))
    assert get(sid)["lines"][0]["quantity"] == "5"
    cancelled = {N.purchasing.requests[k]["request_id"] for k in N.purchasing.requests if N.purchasing.requests[k]["request_id"] in N.purchasing.cancelled}
    open_qty = sum(D(r["quantity"]) for r in N.purchasing.requests.values() if r["request_id"] not in cancelled)
    assert open_qty == D("5"), f"the order sells 5 drop-shipped, purchasing is asked for {open_qty}"


def test_MUST_8_a_free_of_charge_order_amended_above_the_limit_waits_for_a_controller(neighbours, inventory):
    warehouse_set("WH1")
    s = confirm(hand(REP, order_type_code="FREE_OF_CHARGE", lines=[{"product_id": "COLA-075", "quantity": "12", "unit_code": "C62"}]))
    sid = s["sales_order_id"]
    assert s["blocks"] == []                                                                   # 14.40: within the limit
    must(api("POST", f"/sales-orders/{sid}/amend", REP, {"reason": "a bigger sample", "changes": [{"line_no": 1, "quantity": "300"}]}))
    must(api("POST", f"/sales-orders/{sid}/revisions/2/approve", as_role("order_admin"), {}))
    assert D(get(sid)["lines"][0]["statistical_value"]) == D("360.00")
    assert "free_of_charge_approval" in refused(deliver(s, 1, 300, "inv:WH1/OUT/00009", INVENTORY, source="inventory"))


def test_MUST_7_approving_a_free_of_charge_order_never_releases_another_reasons_block(neighbours, inventory):
    admin = as_role("order_admin")
    must(api("POST", "/block-reasons", admin, {"block_reason_code": "CHECK_TERMS", "kind": "delivery", "name": "CHECK_TERMS"}))
    warehouse_set("WH1")
    s = confirm(hand(REP, order_type_code="FREE_OF_CHARGE", lines=[{"product_id": "COLA-075", "quantity": "12", "unit_code": "C62"}]))
    sid = s["sales_order_id"]
    must(api("POST", f"/sales-orders/{sid}/blocks", admin, {"kind": "delivery", "reason_code": "CHECK_TERMS", "note": "terms to check"}))
    api("POST", f"/sales-orders/{sid}/approve-memo", CONTROLLER, {})
    rows = get(sid)["blocks"]
    assert rows[-1]["event"] == "set" and rows[-1]["reason_code"] == "CHECK_TERMS", f"the CHECK_TERMS hold was released by the memo approver: {rows[-1]}"


def test_MUST_1_an_order_amended_below_its_deposit_never_strands_the_customers_deposit(neighbours):
    from test_slice_c import bill, deposit

    from brick_sales_order.application import use_cases
    N = neighbours
    o = confirm(hand(REP, lines=[{"product_id": "FEE", "quantity": "10", "unit_code": "C62"}]))      # 10,000.00, policy ordered
    sid = o["sales_order_id"]
    d1 = deposit(N, o, "5000.00")
    status, _ = api("POST", f"/sales-orders/{sid}/amend", REP, {"reason": "smaller job", "changes": [{"line_no": 1, "quantity": "4"}]})
    if status == 200:
        st, _ = api("POST", f"/sales-orders/{sid}/revisions/2/approve", as_role("order_admin"), {})
        if st != 200:
            return                                                                                   # refused: the deposit stays within the order
    else:
        return
    final = must(bill(o))
    assert final["invoice_requests"][-1]["kind"] == "final"
    with use_cases.repository() as repo:
        used = repo.claims.deducted(d1)
    credited = sum(D(x["net_amount"]) for c in N.invoicing.credits.values() if c["invoice_id"] == d1 for x in c["lines"])
    assert used == D("5000.00") or credited == D("5000.00") - used, f"final invoice billed; deposit 5,000.00 used {used}, credited {credited}: the rest is stranded"


def test_MUST_1_amending_a_return_above_what_was_delivered_is_refused(neighbours, inventory):
    o = original(neighbours, inventory)
    r = confirm(ret(o, "6"))
    rid = r["sales_order_id"]
    st, _ = api("POST", f"/sales-orders/{rid}/amend", REP, {"reason": "more came back", "changes": [{"line_no": 1, "quantity": "30"}]})
    if st == 200:
        api("POST", f"/sales-orders/{rid}/revisions/2/approve", as_role("order_admin"), {})
    assert D(get(rid)["lines"][0]["quantity"]) <= D("24"), "a return of 30 against 24 delivered"
    second = api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "order_type_code": "RETURNS", "return_reason": "DAMAGED",
                                                "lines": [{"product_id": "COLA-075", "quantity": "18"}], "ref_sales_order_id": o["sales_order_id"], "ref_line_no": 1,
                                                "ref_delivery_ref": "inv:WH1/OUT/00001", "ref_invoice_id": "INV-2026-0040", "ref_invoice_line_no": 1})
    total = D(get(rid)["lines"][0]["quantity"]) + (D("18") if second[0] in (200, 201) else 0)
    assert total <= D("24"), f"returns claim {total} of 24 delivered"


def test_MUST_9_a_deposit_whose_invoice_id_is_ars_own_is_still_deducted(neighbours, monkeypatch):
    from test_slice_c import bill
    N = neighbours
    inv = N.invoicing
    real_raise, ids = inv.raise_draft, {}

    def raise_draft(**kw):                                   # AR-03 answers ITS invoice_id; the spec never says it equals the source id
        got = real_raise(**kw)
        ids.setdefault(kw["source_id"], f"AR-{len(ids) + 1:04d}")
        return {**got, "invoice_id": ids[kw["source_id"]]}

    def issued_sources(invoice_ids):                        # AR-30 GET /invoice-totals is keyed by invoice_id
        back = {v: k for k, v in ids.items()}
        return {i: back.get(i) in inv.issued_ids for i in invoice_ids}

    monkeypatch.setattr(inv, "raise_draft", raise_draft)
    monkeypatch.setattr(inv, "issued_sources", issued_sources)
    o = confirm(hand(REP, lines=[{"product_id": "FEE", "quantity": "10", "unit_code": "C62"}]))
    sid = o["sales_order_id"]
    must(api("POST", f"/sales-orders/{sid}/raise-deposit", BILLER, {"amount": "5000.00"}))
    inv.issue(f"so:{sid}:1")
    must(bill(o))
    assert inv.drafts[f"so:{sid}:2"]["deductions"] == [{"deposit_invoice_id": "AR-0001", "net_amount": "5000.00", "tax_category": "standard"}]



def test_MUST_8_a_confirmed_orders_payer_change_is_checked_like_confirmation(neighbours):
    from test_slice_b import EARLIER, WID, partner_rows
    N = neighbours
    N.partners.customers[7312] = True
    partner_rows()
    with on_day(EARLIER):
        must(api("POST", "/customer-partners", as_role("order_admin"), {"sold_to_bp_id": 7311, "function": "payer", "partner_bp_id": 7500, "valid_from": "2026-10-01"}))
    o = confirm(hand(REP, lines=WID))
    N.payments.status[7500] = {"hold": True, "reason": "7500 is over its limit", "mode": "block"}
    st, _ = api("POST", f"/sales-orders/{o['sales_order_id']}/partners", REP, {"payer_bp_id": 7500})
    assert st != 200, "7500 is a supplier on credit hold, yet it now pays for a confirmed order"


def test_MUST_2_a_refused_confirm_leaves_no_billing_schedule_in_invoicing(neighbours, inventory):
    N = neighbours
    warehouse_set("WH1")
    o = hand(REP, lines=[{"product_id": "HOST", "quantity": "12", "unit_code": "MON"}, COLA6])
    inventory.down = True
    assert api("POST", f"/sales-orders/{o['sales_order_id']}/confirm", REP, {})[0] >= 400
    assert get(o["sales_order_id"])["lifecycle"] == "draft"
    assert N.invoicing.schedules == {}, f"invoicing will bill {list(N.invoicing.schedules)} for an order that was never confirmed"


def test_MUST_1_a_draft_changed_into_a_return_takes_the_return_claim(neighbours, inventory):
    o = original(neighbours, inventory)
    ref = {"ref_sales_order_id": o["sales_order_id"], "ref_line_no": 1, "ref_delivery_ref": "inv:WH1/OUT/00001", "ref_invoice_id": "INV-2026-0040",
           "ref_invoice_line_no": 1}
    st, d = api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "lines": [{"product_id": "COLA-075", "quantity": "30", "unit_code": "C62"}], "return_reason": "DAMAGED", **ref})
    if st not in (200, 201):
        return
    st, _ = api("POST", f"/sales-orders/{d['sales_order_id']}/order-type", REP, {"order_type_code": "RETURNS"})
    if st != 200:
        return
    st, _ = api("POST", f"/sales-orders/{d['sales_order_id']}/confirm", REP, {})
    assert st != 200, "a return of 30 against 24 delivered was confirmed through a type change"


def test_MUST_1_a_deposit_is_never_credited_twice_by_a_credit_request_and_the_cancel(neighbours):
    from test_slice_c import deposit
    N = neighbours
    o = confirm(hand(REP, lines=[{"product_id": "FEE", "quantity": "10", "unit_code": "C62"}]))
    sid = o["sales_order_id"]
    d1 = deposit(N, o, "5000.00")
    N.invoicing.add_billed(d1, 7311, "2026-11-20", [{"net_amount": "5000.00", "product_id": "FEE"}])      # ar lists the issued deposit invoice's line
    cr = confirm(hand(REP, order_type_code="CREDIT_REQ", ref_invoice_id=d1, return_reason="PRICE_ERROR",
                      lines=[{"product_id": "FEE", "quantity": "1", "unit_code": "C62", "unit_price": "5000.00", "ref_invoice_line_no": 1}]))
    must(api("POST", f"/sales-orders/{cr['sales_order_id']}/approve-memo", CONTROLLER, {}))
    api("POST", f"/sales-orders/{sid}/cancel", as_role("order_admin"), {"reason": "the customer withdrew"})
    credited = sum(D(x["net_amount"]) for c in N.invoicing.credits.values() if c["invoice_id"] == d1 for x in c["lines"])
    assert credited <= D("5000.00"), f"a 5,000.00 deposit was credited {credited}"


def test_a_send_that_failed_after_the_commit_is_completed_by_the_next_daily_run_exactly_once(neighbours, inventory, monkeypatch):
    from test_slice_b import reschedule
    N = neighbours
    sent = []
    real = N.purchasing.purchase_request
    monkeypatch.setattr(N.purchasing, "purchase_request", lambda **kw: (sent.append(kw["request_ref"]), real(**kw))[1])
    N.purchasing.down = True
    o = confirm(hand(REP, ship_to_bp_id=7313, lines=PRINTER))                         # the order commits; purchasing did not answer
    sid = o["sales_order_id"]
    assert o["lines"][0]["purchase_request_id"] == "" and get(sid)["lifecycle"] == "confirmed"
    N.purchasing.down = False
    reschedule()                                                                     # the daily run completes the owed send ...
    assert get(sid)["lines"][0]["purchase_request_id"] == N.purchasing.requests[f"so:{sid}:1@r1"]["request_id"]
    n = len(sent)
    reschedule()                                                                     # ... and a second run sends nothing
    assert len(sent) == n and len(N.purchasing.requests) == 1


def test_a_cancel_purchasing_refuses_after_inventory_released_the_stock_holds_it_again_exactly_once(neighbours, inventory):
    from test_slice_b import reschedule
    N = neighbours
    warehouse_set("WH1")
    o = confirm(hand(REP, ship_to_bp_id=7313, lines=[*PRINTER, COLA6]))
    sid, rid = o["sales_order_id"], o["lines"][0]["purchase_request_id"]
    N.purchasing.confirmed_with_supplier[rid] = "PO-0144 is confirmed with the supplier"
    assert "third_party_cancellable" in refused(api("POST", f"/sales-orders/{sid}/cancel", as_role("order_admin"), {"reason": "changed our mind"}))
    assert get(sid)["lifecycle"] == "confirmed" and inventory.released == [sid]
    from brick_sales_order.application import use_cases
    with use_cases.repository() as repo:
        assert [p["event"] for p in repo.postings.pending_reservations() if p["document_id"].startswith(sid)]
    before = len([c for c in inventory.calls if c[0] == "reserve" and c[1] == sid])
    reschedule()
    after = len([c for c in inventory.calls if c[0] == "reserve" and c[1] == sid])
    assert after == before + 1
    reschedule()
    assert len([c for c in inventory.calls if c[0] == "reserve" and c[1] == sid]) == after