dodil/erp-brick-sales-order
DODIL ERP brick: sales order
tests/test_slice_c.py
"""Slice C of sales-order v6: returns and their credit (SO-41, SO-49), credit and debit memo requests (SO-42), free of charge (SO-43), RUSH and CASH
(SO-44), consignment (SO-45), milestones (SO-46), deposits deducted per deposit (SO-47) and third-party lines (SO-48)."""
import json
from decimal import Decimal as D
from conftest import (
BILLER,
FULFILLER,
INVENTORY,
PURCHASING,
REP,
SYSTEM,
api,
as_role,
confirm,
deliver,
get,
hand,
must,
on_day,
person,
refused,
)
from erp_platform.repository import CLAIM_TABLES
from test_slice_b import setting, warehouse_set
from brick_sales_order.application import use_cases
from brick_sales_order.infrastructure import tables
CONTROLLER = as_role("sales_controller")
COLA24 = [{"product_id": "COLA-075", "quantity": "24", "unit_code": "C62"}]
INV40 = "INV-2026-0040"
def _with_sub(headers, sub):
user = json.loads(headers["X-Dodil-User"])
return {**headers, "X-Dodil-User": json.dumps({**user, "sub": sub})}
def approver(email, sub=None):
return _with_sub(person(email, ["so:order:read", "so:sales_order:read_all", "so:memo:approve"]), sub or email)
def original(N, inventory):
"""SO-2026-0012 of the card: 24 C62 of COLA-075 at 1.20, delivered by inventory (inv:WH1/OUT/00001) and invoiced on INV-2026-0040 (28.80 net)."""
warehouse_set("WH1")
o = confirm(hand(REP, lines=COLA24))
must(deliver(o, 1, 24, "inv:WH1/OUT/00001", INVENTORY, source="inventory"))
N.invoicing.add_billed(INV40, 7311, "2026-11-20", [{"net_amount": "28.80", "so_line_no": 1, "sales_order_id": o["sales_order_id"]}])
return o
def ret(o, qty="6", reason="DAMAGED", **kw):
ref = {"ref_sales_order_id": o["sales_order_id"], "ref_line_no": 1, "ref_delivery_ref": "inv:WH1/OUT/00001", "ref_invoice_id": INV40, "ref_invoice_line_no": 1}
return hand(REP, order_type_code="RETURNS", lines=[{"product_id": "COLA-075", "quantity": qty}], **ref, return_reason=reason, **kw)
def receive(r, qty, ref="inv:WH1/RET/00001"):
return must(deliver(r, 1, qty, ref, INVENTORY, source="inventory"))
def inspect(r, accepted, rejected, ref="inv:RET-1:1", received=None, **kw):
return api("POST", f"/sales-orders/{r['sales_order_id']}/return-inspections", INVENTORY,
{"line_no": 1, "inspection_ref": ref, "received_quantity": str(received or accepted + rejected), "accepted_quantity": str(accepted),
"rejected_quantity": str(rejected), "decision": "restock", "reason": kw.get("reason", "1 scrapped (dented); 1 rejected: opened"),
"inspected_on": "2026-11-20"})
# ── SO-41 returns ─────────────────────────────────────────────────────────────────────────────
def test_SO_41_a_return_is_its_own_order_inspected_and_credited_once_against_the_original_invoice(neighbours, inventory):
N, o = neighbours, original(neighbours, inventory)
too_many = refused(api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "order_type_code": "RETURNS", "return_reason": "DAMAGED",
"lines": [{"product_id": "COLA-075", "quantity": "30"}], "ref_sales_order_id": o["sales_order_id"], "ref_line_no": 1,
"ref_delivery_ref": "inv:WH1/OUT/00001", "ref_invoice_id": INV40, "ref_invoice_line_no": 1}))
assert "within_delivered" in too_many and "24 delivered, 0 returned: at most 24" in too_many
d = ret(o)
line = d["lines"][0]
assert (line["unit_code"], D(line["unit_price"]), line["tax_category"], line["ref_invoice_id"], line["ref_delivery_ref"], line["ref_line_no"]) == \
("C62", D("1.20"), "standard", INV40, "inv:WH1/OUT/00001", 1)
r = confirm(d)
assert r["order_no"] == "RE-2026-0001" and r["lifecycle"] == "confirmed"
exp = inventory.expected[f"so:{r['sales_order_id']}"]
assert exp["warehouse_id"] == "WH1" and exp["lines"][0]["original_delivery_ref"] == "inv:WH1/OUT/00001" and exp["lines"][0]["inspect"] is True
assert not [c for c in inventory.calls if c[0] == "reserve" and c[1] == r["sales_order_id"]] # nothing is reserved for goods coming back
assert "billing_relevant" in refused(api("POST", f"/sales-orders/{r['sales_order_id']}/request-invoice", BILLER, {}))
receive(r, 6)
assert not N.invoicing.credits # return_credit_on inspection: not yet
got = must(inspect(r, 5, 1))
assert got["line"]["received"] == "6" and got["line"]["accepted"] == "5" and got["line"]["rejected"] == "1" and got["line"]["credited"] == "5"
assert got["line"]["rejected_reasons"][0]["reason"].endswith("rejected: opened")
credit = N.invoicing.credits[f"so:{r['sales_order_id']}:1"]
assert credit["invoice_id"] == INV40 and [(x["invoice_line_no"], x["quantity"], x["unit_price"], x["net_amount"], x["so_line_no"], x["sales_order_id"])
for x in credit["lines"]] == [(1, "5", "1.20", "6.00", 1, r["sales_order_id"])]
assert get(r["sales_order_id"])["lifecycle"] == "done" # every line inspected and credited: closed
orig = get(o["sales_order_id"]) # a return never changes the original order
assert orig["lines"][0]["delivered"] == "24" and orig["invoice_requests"] == []
n = len(N.invoicing.credits)
must(inspect(r, 5, 1)) # the same inspection again: recorded once, credited once
assert len(N.invoicing.credits) == n and len(get(r["sales_order_id"])["inspections"]) == 1
assert "within_received" in refused(inspect(r, 1, 0, ref="inv:RET-1:2")) # 6 received, 6 inspected: no more
def test_SO_41_a_reason_with_a_restocking_fee_and_credit_on_receipt(neighbours, inventory):
N, o = neighbours, original(neighbours, inventory)
r = confirm(ret(o, "6", "CHANGED_MIND"))
receive(r, 6)
assert not N.invoicing.credits
must(inspect(r, 6, 0))
assert [x["net_amount"] for x in N.invoicing.credits[f"so:{r['sales_order_id']}:1"]["lines"]] == ["6.48"] # 7.20 less 10%
setting("return_credit_on", "receipt")
r2 = confirm(ret(o, "6"))
receive(r2, 6) # return_credit_on receipt: credited when received
assert [x["net_amount"] for x in N.invoicing.credits[f"so:{r2['sales_order_id']}:1"]["lines"]] == ["7.20"]
assert "within_delivered" in refused(api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "order_type_code": "RETURNS", "return_reason": "DAMAGED",
"lines": [{"product_id": "COLA-075", "quantity": "13"}], "ref_sales_order_id": o["sales_order_id"],
"ref_line_no": 1, "ref_delivery_ref": "inv:WH1/OUT/00001", "ref_invoice_id": INV40, "ref_invoice_line_no": 1}))
def test_SO_41_a_credit_that_could_not_be_sent_stays_pending_and_the_retry_sends_it_once(neighbours, inventory):
N, o = neighbours, original(neighbours, inventory)
r = confirm(ret(o))
receive(r, 6)
N.invoicing.credit_down = True
assert "invoicing did not answer" in refused(inspect(r, 6, 0), 502)
pending = get(r["sales_order_id"])["invoice_requests"]
assert [(x["kind"], x["status"]) for x in pending] == [("credit", "pending")] and not N.invoicing.credits
N.invoicing.credit_down = False
must(inspect(r, 6, 0))
done = get(r["sales_order_id"])["invoice_requests"]
assert [(x["kind"], x["status"], x["request_no"]) for x in done] == [("credit", "requested", 1)] and len(N.invoicing.credits) == 1
def test_SO_41_the_return_claim_is_released_by_cancelling_and_a_return_needs_an_open_rma(neighbours, inventory):
o = original(neighbours, inventory)
r = confirm(ret(o, "24"))
assert "within_delivered" in refused(api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "order_type_code": "RETURNS", "return_reason": "DAMAGED",
"lines": [{"product_id": "COLA-075", "quantity": "1"}], "ref_sales_order_id": o["sales_order_id"],
"ref_line_no": 1, "ref_delivery_ref": "inv:WH1/OUT/00001", "ref_invoice_id": INV40, "ref_invoice_line_no": 1}))
must(api("POST", f"/sales-orders/{r['sales_order_id']}/cancel-return", REP, {"reason": "the customer keeps them"}))
assert inventory.cancelled_returns == [f"so:{r['sales_order_id']}"]
again = confirm(ret(o, "24")) # the quantity is free again
assert again["lifecycle"] == "confirmed"
with use_cases.repository() as repo:
assert repo.claims.returned(f"ret:{o['sales_order_id']}:1") == D("24") and "so_credit_claim" in CLAIM_TABLES
# ── SO-42 credit and debit memo requests ──────────────────────────────────────────────────────
def memo(N, inv_line=1, price="0.10", qty="24", **kw):
return hand(REP, order_type_code="CREDIT_REQ", ref_invoice_id=INV40, return_reason="PRICE_ERROR", **kw,
lines=[{"product_id": "COLA-075", "quantity": qty, "unit_code": "C62", "unit_price": price, "ref_invoice_line_no": inv_line}])
def test_SO_42_a_credit_and_a_debit_request_are_approved_by_a_second_person_and_reach_invoicing_keyed(neighbours, inventory):
N, o = neighbours, original(neighbours, inventory)
cr = confirm(memo(N))
dr = confirm(hand(REP, order_type_code="DEBIT_REQ", ref_invoice_id=INV40, lines=[{"product_id": "FEE", "quantity": "1", "unit_price": "15.00",
"description": "freight forgotten"}]))
assert (cr["order_no"], dr["order_no"]) == ("CR-2026-0001", "DR-2026-0001")
assert [b["reason_code"] for b in cr["blocks"]] == ["APPROVAL", "APPROVAL"] and D(cr["total"]) > 0
assert not [c for c in inventory.calls if c[0] == "reserve" and c[1] == cr["sales_order_id"]] # neither is delivery-relevant
cid = cr["sales_order_id"]
assert api("POST", f"/sales-orders/{cid}/approve-memo", REP, {})[0] == 403
same_sub = approver("rep.second.account@shop.example", sub="sales_rep@shop.example")
same_email = approver("sales_rep@shop.example", sub="another-sub")
for who in (same_sub, same_email):
assert "approver_not_requester" in refused(api("POST", f"/sales-orders/{cid}/approve-memo", who, {}))
assert not N.invoicing.credits and get(cid)["lifecycle"] == "confirmed"
done = must(api("POST", f"/sales-orders/{cid}/approve-memo", CONTROLLER, {}))
credit = N.invoicing.credits[f"so:{cid}:1"]
assert credit["invoice_id"] == INV40 and credit["reason"].endswith("PRICE_ERROR")
assert [(x["invoice_line_no"], x["quantity"], x["unit_price"], x["net_amount"], x["so_line_no"], x["sales_order_id"]) for x in credit["lines"]] == \
[(1, "24", "0.10", "2.40", 1, cid)]
assert done["lifecycle"] == "done" and done["approved_by_email"] == "sales_controller@shop.example" and all(b["event"] == "release" for b in done["blocks"][2:])
must(api("POST", f"/sales-orders/{dr['sales_order_id']}/approve-memo", CONTROLLER, {}))
draft = N.invoicing.drafts[f"so:{dr['sales_order_id']}:1"]
assert draft["kind"] == "regular" and INV40 in draft["lines"][0]["description"] and draft["lines"][0]["net_amount"] == "15.00"
assert get(o["sales_order_id"])["invoice_requests"] == [] # the original order's billed quantity is untouched
n = len(N.invoicing.credits)
must(api("POST", f"/sales-orders/{cid}/approve-memo", CONTROLLER, {})) # approving twice credits once
assert len(N.invoicing.credits) == n
def test_SO_42_every_credit_on_an_invoice_line_shares_one_claim_so_the_last_pound_goes_once(neighbours, inventory):
N, o = neighbours, original(neighbours, inventory)
first = confirm(memo(N)) # 2.40 of the 28.80
must(api("POST", f"/sales-orders/{first['sales_order_id']}/approve-memo", CONTROLLER, {}))
r = confirm(ret(o, "6")) # a return's credit of 7.20 takes from the same line
receive(r, 6)
must(inspect(r, 6, 0))
left = D("28.80") - D("2.40") - D("7.20") # 19.20
big = confirm(memo(N, price="0.80")) # 24 x 0.80 = 19.20 exactly what is left
huge = confirm(memo(N, price="0.81")) # 19.44: one pound-ish too many
said = refused(api("POST", f"/sales-orders/{huge['sales_order_id']}/approve-memo", CONTROLLER, {}))
assert "within_invoiced" in said and f"{left} is left to credit" in said
assert not any(k.startswith(f"so:{huge['sales_order_id']}") for k in N.invoicing.credits) and get(huge["sales_order_id"])["lifecycle"] == "confirmed"
must(api("POST", f"/sales-orders/{big['sales_order_id']}/approve-memo", CONTROLLER, {}))
again = confirm(memo(N, price="0.01"))
assert "0.00 is left" in refused(api("POST", f"/sales-orders/{again['sales_order_id']}/approve-memo", CONTROLLER, {}))
with use_cases.repository() as repo: # a racer that read the same seq is refused at the insert, never a second claim
key = f"inv:{INV40}:1"
seq = repo.claims._next(tables.so_credit_claim, "claim_key", key)
repo.claims.claim(key, "credit", "X", amount=D("0.01"))
try:
repo.conn.execute(tables.so_credit_claim.insert().values(claim_key=key, seq=seq, kind="credit", sales_order_id="Y", amount=D("0.01")))
raced = False
except Exception as e: # noqa: BLE001 — the engine's duplicate key
raced = "UNIQUE" in str(e) or "unique" in str(e)
assert raced
# ── SO-43 free of charge ──────────────────────────────────────────────────────────────────────
def test_SO_43_samples_are_priced_for_statistics_never_billed_and_a_replacement_above_the_limit_waits_for_a_controller(neighbours, inventory):
N = neighbours
s = hand(REP, order_type_code="FREE_OF_CHARGE", lines=[{"product_id": "COLA-075", "quantity": "12", "unit_code": "C62"}])
assert (D(s["lines"][0]["net_amount"]), D(s["lines"][0]["statistical_value"]), s["lines"][0]["item_category_code"]) == (D("0.00"), D("14.40"), "FREE")
assert D(s["total"]) == 0
s = confirm(s)
assert s["order_no"] == "FC-2026-0001" and s["blocks"] == [] # 14.40 is within the limit of 250.00
assert "billing_relevant" in refused(api("POST", f"/sales-orders/{s['sales_order_id']}/request-invoice", BILLER, {}))
fulfil = must(api("GET", f"/sales-orders/{s['sales_order_id']}/fulfilment", INVENTORY))
assert fulfil["purpose"] == "free_of_charge"
out = must(deliver(s, 1, 12, "inv:WH1/OUT/00002", INVENTORY, source="inventory"))
assert out["lifecycle"] == "done" and out["invoice_status"] == "nothing to invoice" and not N.invoicing.drafts
rep = confirm(hand(REP, order_type_code="FREE_OF_CHARGE", ref_sales_order_id="RE-2026-0002", lines=[{"product_id": "DRILL-500", "quantity": "1", "unit_code": "C62"}]))
rid = rep["sales_order_id"]
assert D(rep["lines"][0]["statistical_value"]) == D("320.00") and [b["reason_code"] for b in rep["blocks"]] == ["APPROVAL", "APPROVAL"]
gates = {g["gate"]: g for g in must(api("GET", f"/sales-orders/{rid}/readiness", REP))["readiness"]["delivery"]}
assert not gates["free_of_charge_approval"]["met"] and "320.00 is above the limit of 250.00" in gates["free_of_charge_approval"]["reason"]
assert "free_of_charge_approval" in refused(deliver(rep, 1, 1, "inv:WH1/OUT/00003", INVENTORY, source="inventory"))
assert api("POST", f"/sales-orders/{rid}/approve-memo", REP, {})[0] == 403
assert "approver_not_requester" in refused(api("POST", f"/sales-orders/{rid}/approve-memo", approver("rep.alt@shop.example", "sales_rep@shop.example"), {}))
must(api("POST", f"/sales-orders/{rid}/approve-memo", CONTROLLER, {}))
out = must(deliver(rep, 1, 1, "inv:WH1/OUT/00003", INVENTORY, source="inventory"))
assert out["lifecycle"] == "done" and not N.invoicing.drafts
assert "nothing waiting for approval" in refused(api("POST", f"/sales-orders/{s['sales_order_id']}/approve-memo", CONTROLLER, {}))
# ── SO-44 RUSH and CASH ───────────────────────────────────────────────────────────────────────
def test_SO_44_a_rush_order_asks_inventory_for_a_delivery_today_with_priority_rush(neighbours, inventory):
warehouse_set("WH1")
o = confirm(hand(REP, order_type_code="RUSH", lines=[{"product_id": "COLA-075", "quantity": "6", "unit_code": "C62"}]))
assert o["priority"] == "rush" and o["requested_delivery_on"] == "2026-11-20" and o["order_no"].startswith("RU-2026-")
body = inventory.reserve_bodies[-1][1]
assert body["priority"] == "rush" and body["schedule_version"] >= 1 and body["lines"][0]["warehouse_id"] == "WH1"
assert must(api("GET", f"/sales-orders/{o['sales_order_id']}/fulfilment", INVENTORY))["priority"] == "rush"
plain = confirm(hand(REP, lines=[{"product_id": "COLA-075", "quantity": "6", "unit_code": "C62"}]))
assert "priority" not in inventory.reserve_bodies[-1][1] or inventory.reserve_bodies[-1][1]["priority"] == "normal"
assert plain["priority"] == "normal"
def test_SO_44_a_cash_order_is_not_credit_checked_and_is_invoiced_and_issued_in_the_same_step(neighbours, inventory):
N = neighbours
N.payments.status[8420] = {"hold": True, "reason": "walk-in on hold", "mode": "block"}
lines = [{"product_id": "COLA-075", "quantity": "6", "unit_code": "C62"}, {"product_id": "SETUP", "quantity": "1", "unit_code": "HUR"}]
d = hand(REP, bp_id=8420, order_type_code="CASH", lines=lines)
o = confirm(d) # the credit gate is skipped: paid now
sid = o["sales_order_id"]
assert o["order_no"].startswith("CA-2026-")
draft = N.invoicing.drafts[f"so:{sid}:1"]
assert draft["issue_hint"] == "cash" and f"so:{sid}:1" in N.invoicing.issued_ids and [x["line_no"] for x in draft["lines"]] == [1] # the service line bills when done
must(deliver(o, 1, 6, "inv:WH1/OUT/00009", INVENTORY, source="inventory")) # validated at the counter
must(deliver(o, 2, 1, "DN-1")) # a service line is recorded by hand
final = must(api("POST", f"/sales-orders/{sid}/request-invoice", BILLER, {}))
assert f"so:{sid}:2" in N.invoicing.issued_ids and N.invoicing.drafts[f"so:{sid}:2"]["issue_hint"] == "cash"
assert must(api("POST", f"/sales-orders/{sid}/finish", SYSTEM, {}))["lifecycle"] == "done" and final["invoice_status"] == "invoiced"
assert "cash_no_deposit" in refused(api("POST", f"/sales-orders/{sid}/raise-deposit", BILLER, {"amount": "1.00"}))
plan = hand(REP, bp_id=8420, order_type_code="CASH", lines=[{"product_id": "HOST", "quantity": "12", "unit_code": "MON"}])
assert "cash_no_plan" in refused(api("POST", f"/sales-orders/{plan['sales_order_id']}/confirm", REP, {}))
# ── SO-45 consignment ─────────────────────────────────────────────────────────────────────────
def test_SO_45_consignment_is_filled_issued_at_the_issue_dates_price_picked_up_and_returned_and_the_stock_is_inventorys(neighbours, inventory):
N = neighbours
warehouse_set("WH1")
inventory.consignment[(7311, "COLA-075")] = D("240")
fill = confirm(hand(REP, order_type_code="CONS_FILL", lines=[{"product_id": "COLA-075", "quantity": "240", "unit_code": "C62"}]))
fid = fill["sales_order_id"]
assert fill["order_no"] == "CS-2026-0001" and D(fill["lines"][0]["net_amount"]) == 0 and D(fill["lines"][0]["statistical_value"]) == D("288.00")
f = must(api("GET", f"/sales-orders/{fid}/fulfilment", INVENTORY))
assert (f["special_stock"], f["purpose"]) == ("to_consignment", "consignment")
assert "billing_relevant" in refused(api("POST", f"/sales-orders/{fid}/request-invoice", BILLER, {})) # a fill is never invoiced
assert must(deliver(fill, 1, 240, "inv:WH1/CONS/00001", INVENTORY, source="inventory"))["lifecycle"] == "done" and not N.invoicing.drafts
with on_day("2026-12-02T09:00:00+00:00"): # the price changed on 2026-12-01
issue = confirm(hand(REP, order_type_code="CONS_ISSUE", ref_sales_order_id=fid, lines=[{"product_id": "COLA-075", "quantity": "60", "unit_code": "C62"}]))
iid = issue["sales_order_id"]
assert D(issue["lines"][0]["unit_price"]) == D("1.30") and issue["lines"][0]["priced_on"] == "2026-12-02" # priced on the issue date
must(deliver(issue, 1, 60, "inv:WH1/ISS/00001", INVENTORY, source="inventory"))
must(api("POST", f"/sales-orders/{iid}/request-invoice", BILLER, {}))
assert [x["net_amount"] for x in N.invoicing.drafts[f"so:{iid}:1"]["lines"]] == ["78.00"]
N.invoicing.issue(f"so:{iid}:1")
N.invoicing.add_billed(f"so:{iid}:1", 7311, "2026-12-02", [{"net_amount": "78.00", "so_line_no": 1, "sales_order_id": iid}])
inventory.consignment[(7311, "COLA-075")] = D("180") # inventory moved 60 to the customer: 180 left
pick = confirm(hand(REP, order_type_code="CONS_PICKUP", ref_sales_order_id=fid, lines=[{"product_id": "COLA-075", "quantity": "24", "unit_code": "C62"}]))
assert "billing_relevant" in refused(api("POST", f"/sales-orders/{pick['sales_order_id']}/request-invoice", BILLER, {})) # a pick-up bills nothing
inventory.consignment[(7311, "COLA-075")] = D("156")
back = confirm(hand(REP, order_type_code="CONS_RETURN", ref_sales_order_id=iid, ref_line_no=1, ref_delivery_ref="inv:WH1/ISS/00001", ref_invoice_id=f"so:{iid}:1",
ref_invoice_line_no=1, lines=[{"product_id": "COLA-075", "quantity": "6"}]))
must(deliver(back, 1, 6, "inv:WH1/RET/00001", INVENTORY, source="inventory")) # received: credited against the issue's invoice (a return)
assert [(x["invoice_line_no"], x["quantity"], x["net_amount"]) for x in N.invoicing.credits[f"so:{back['sales_order_id']}:1"]["lines"]] == [(1, "6", "7.80")]
inventory.consignment[(7311, "COLA-075")] = D("162")
said = refused(api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "order_type_code": "CONS_ISSUE", "ref_sales_order_id": fid,
"lines": [{"product_id": "COLA-075", "quantity": "200", "unit_code": "C62"}]}))
assert "consignment_on_hand" in said and "7311 holds 162 C62 of COLA-075 on consignment" in said
inventory.consignment[(7311, "COLA-075")] = D("100") # read from inventory every time, never kept here
draft = hand(REP, order_type_code="CONS_ISSUE", ref_sales_order_id=fid, lines=[{"product_id": "COLA-075", "quantity": "60", "unit_code": "C62"}])
inventory.consignment[(7311, "COLA-075")] = D("30")
assert "consignment_on_hand" in refused(api("POST", f"/sales-orders/{draft['sales_order_id']}/confirm", REP, {}))
# ── SO-46 milestones ──────────────────────────────────────────────────────────────────────────
def stages(*pcts):
names = ["on order", "installed", "accepted", "final"]
events = ["confirmation", "installed", "accepted", "final"]
return [{"pct": p, "event": events[i], "name": names[i]} for i, p in enumerate(pcts)]
def milestone_order(*pcts, price_product="PROJ-INSTALL", extra=()):
return hand(REP, lines=[{"product_id": price_product, "quantity": "1", "unit_code": "C62", "milestones": stages(*pcts)}, *extra])
def release(o, line, no, who=FULFILLER):
return api("POST", f"/sales-orders/{o['sales_order_id']}/milestones/{line}/{no}/release", who, {})
def bill(o):
return api("POST", f"/sales-orders/{o['sales_order_id']}/request-invoice", BILLER, {})
def test_SO_46_a_milestone_line_is_billed_by_value_each_stage_on_its_release_and_the_last_takes_the_remainder(neighbours):
N = neighbours
o = milestone_order("30", "40", "30", extra=[{"product_id": "HOST", "quantity": "12", "unit_code": "MON"}])
bad = milestone_order("30", "40", "20")
assert "plan_complete" in refused(api("POST", f"/sales-orders/{bad['sales_order_id']}/confirm", REP, {})) and "total 90%" in \
refused(api("POST", f"/sales-orders/{bad['sales_order_id']}/confirm", REP, {}))
o = confirm(o)
sid = o["sales_order_id"]
assert list(N.invoicing.schedules) == [f"{sid}|2"] # the monthly line is invoicing's schedule
must(bill(o))
m1 = N.invoicing.drafts[f"so:{sid}:1"]["lines"]
assert [(x["description"], x["net_amount"], x["so_line_no"] if "so_line_no" in x else x["line_no"]) for x in m1] == [("Milestone 1 of 3: on order (30%)", "6000.00", 1)]
assert "something_due" in refused(bill(o)) # M2 holds until it is released
assert api("POST", f"/sales-orders/{sid}/milestones/1/2/release", REP, {})[0] == 403 # the fulfiller releases
must(release(o, 1, 2))
assert "milestone_open" in refused(release(o, 1, 2)) # a stage releases once
must(bill(o))
assert [x["net_amount"] for x in N.invoicing.drafts[f"so:{sid}:2"]["lines"]] == ["8000.00"]
must(release(o, 1, 3))
last = must(bill(o))
assert [x["net_amount"] for x in N.invoicing.drafts[f"so:{sid}:3"]["lines"]] == ["6000.00"] and N.invoicing.drafts[f"so:{sid}:3"]["kind"] == "final"
assert last["invoice_status"] == "invoiced" and sum(D(x["amount"]) for x in last["invoice_requests"]) == D("20000.00")
def test_SO_46_stages_that_do_not_divide_evenly_still_sum_exactly_and_an_amendment_respreads_only_what_is_unreleased(neighbours):
N = neighbours
o = confirm(milestone_order("33.3333", "33.3333", "33.3334", price_product="FEE"))
sid = o["sales_order_id"]
must(bill(o))
must(release(o, 1, 2))
must(bill(o))
must(release(o, 1, 3))
must(bill(o))
billed = [D(N.invoicing.drafts[f"so:{sid}:{n}"]["lines"][0]["net_amount"]) for n in (1, 2, 3)]
assert billed == [D("333.33"), D("333.33"), D("333.34")] and sum(billed) == D("1000.00")
p = confirm(milestone_order("30", "40", "30"))
pid = p["sales_order_id"]
must(bill(p)) # M1: 6,000.00, frozen when released
must(api("POST", f"/sales-orders/{pid}/amend", REP, {"reason": "the scope doubled", "changes": [{"line_no": 1, "quantity": "2"}]}))
must(api("POST", f"/sales-orders/{pid}/revisions/2/approve", as_role("order_admin"), {}))
must(release(p, 1, 2))
must(bill(p))
assert [x["net_amount"] for x in N.invoicing.drafts[f"so:{pid}:2"]["lines"]] == ["19428.57"] # (40,000 - 6,000) x 40 / 70
must(release(p, 1, 3))
must(bill(p))
assert [x["net_amount"] for x in N.invoicing.drafts[f"so:{pid}:3"]["lines"]] == ["14571.43"] # the rest: 40,000 - 6,000 - 19,428.57
def test_SO_46_equal_instalments_of_a_fixed_total_go_to_invoicing_as_a_schedule(neighbours):
N = neighbours
o = confirm(hand(REP, lines=[{"product_id": "FEE", "quantity": "1", "unit_code": "C62", "instalments": 4}]))
sch = N.invoicing.schedules[f"{o['sales_order_id']}|1"]
assert (sch["billing_kind"], sch["quantity"], sch["unit_price"]) == ("instalment", "4", "250.00")
odd = hand(REP, lines=[{"product_id": "FEE", "quantity": "1", "unit_code": "C62", "instalments": 3}])
assert "equal instalments to the penny" in refused(api("POST", f"/sales-orders/{odd['sales_order_id']}/confirm", REP, {}), 422)
# ── SO-47 deposits deducted per deposit ───────────────────────────────────────────────────────
WIDGETS1000 = [{"product_id": "WIDGET", "quantity": "1000", "unit_code": "C62"}] # 10,000.00 net, delivered policy, standard rated
def deposit(N, o, amount, issue=True):
sid = o["sales_order_id"]
out = must(api("POST", f"/sales-orders/{sid}/raise-deposit", BILLER, {"amount": amount}))
req = out["invoice_requests"][-1]
if issue:
N.invoicing.issue(f"so:{sid}:{req['request_no']}") # invoicing issues the draft it holds under our source id ...
return req["invoice_id"] # ... and ar's own invoice_id (AR-0001) is what the order keeps and credits against
def mode_order(mode):
from test_stories import set_policy
set_policy(7311, "2026-11-20", "2026-11-19", deposit_deduction_mode=mode)
return confirm(hand(REP, lines=WIDGETS1000))
def test_SO_47_a_deposit_is_deducted_per_deposit_oldest_first_in_its_tax_category_and_the_final_takes_what_is_left(neighbours):
N = neighbours
o = confirm(hand(REP, lines=WIDGETS1000)) # deposit mode proportional, the order's default
sid = o["sales_order_id"]
assert o["deposit_deduction_mode"] == "proportional"
d1, d2 = deposit(N, o, "5000.00"), deposit(N, o, "1000.00")
must(deliver(o, 1, 400, "DN-1"))
first = must(bill(o))
req = first["invoice_requests"][2]
assert (req["kind"], D(req["deducted"]), D(req["amount"])) == ("regular", D("2400.00"), D("1600.00")) # 6,000 / 10,000 x 4,000
assert N.invoicing.drafts[f"so:{sid}:3"]["deductions"] == [{"deposit_invoice_id": d1, "net_amount": "2400.00", "tax_category": "standard"}]
must(deliver(o, 1, 600, "DN-2"))
final = must(bill(o))
req = final["invoice_requests"][3]
assert (req["kind"], D(req["deducted"]), D(req["amount"])) == ("final", D("3600.00"), D("2400.00"))
assert N.invoicing.drafts[f"so:{sid}:4"]["deductions"] == [{"deposit_invoice_id": d1, "net_amount": "2600.00", "tax_category": "standard"},
{"deposit_invoice_id": d2, "net_amount": "1000.00", "tax_category": "standard"}]
with use_cases.repository() as repo:
assert (repo.claims.deducted(d1), repo.claims.deducted(d2)) == (D("5000.00"), D("1000.00")) and "so_deposit_deduction" in CLAIM_TABLES
seq = repo.claims._next(tables.so_deposit_deduction, "deposit_invoice_id", d1)
repo.claims.claim_deposit(d1, "deduct", "X", 9, D("0.01"), "standard")
try: # a racer that read the same seq is refused at the insert
repo.conn.execute(tables.so_deposit_deduction.insert().values(deposit_invoice_id=d1, seq=seq, kind="deduct", sales_order_id="Y", amount=D("0.01")))
raced = False
except Exception as e: # noqa: BLE001 — the engine's duplicate key
raced = "unique" in str(e).lower()
assert raced
def test_SO_47_full_first_manual_and_a_draft_deposit(neighbours):
N = neighbours
o = mode_order("full_first")
sid = o["sales_order_id"]
d1 = deposit(N, o, "5000.00")
must(deliver(o, 1, 400, "DN-1"))
req = must(bill(o))["invoice_requests"][1]
assert (D(req["deducted"]), D(req["amount"])) == (D("4000.00"), D("0.00")) # as much as it can: all of this invoice
assert N.invoicing.drafts[f"so:{sid}:2"]["deductions"][0]["net_amount"] == "4000.00" and D(req["amount"]) == 0
with use_cases.repository() as repo:
assert repo.claims.deducted(d1) == D("4000.00")
from test_stories import set_policy
set_policy(7313, "2026-11-20", "2026-11-19", deposit_deduction_mode="manual")
m = confirm(hand(REP, bp_id=7313, lines=WIDGETS1000))
msid = m["sales_order_id"]
issued, draft = deposit(N, m, "3000.00"), deposit(N, m, "500.00", issue=False)
must(deliver(m, 1, 400, "DN-1"))
def body(*ds):
return api("POST", f"/sales-orders/{msid}/request-invoice", BILLER, {"deductions": [{"deposit_invoice_id": i, "amount": a} for i, a in ds]})
assert "deposit_remaining" in refused(body((issued, "3000.01"))) and "3000.00 left" in refused(body((issued, "3000.01")))
assert "still a draft" in refused(body((draft, "100.00"))) or "not an issued deposit" in refused(body((draft, "100.00")))
got = must(body((issued, "1500.00")))["invoice_requests"][2]
assert (D(got["deducted"]), D(got["amount"])) == (D("1500.00"), D("2500.00"))
must(deliver(m, 1, 200, "DN-2"))
assert "deposit_remaining" in refused(body((issued, "1500.01"))) and "1500.00 left" in refused(body((issued, "1500.01"))) # 1,500.00 is left, not 3,000.00
def test_SO_47_cancelling_an_order_with_deposit_left_credits_the_deposit_invoice(neighbours):
N = neighbours
o = confirm(hand(REP, lines=WIDGETS1000))
sid = o["sales_order_id"]
d1 = deposit(N, o, "5000.00")
out = must(api("POST", f"/sales-orders/{sid}/cancel", as_role("order_admin"), {"reason": "the customer withdrew"}))
credit = N.invoicing.credits[f"so:{sid}:2"]
assert out["lifecycle"] == "cancelled" and credit["invoice_id"] == d1 and [(x["invoice_line_no"], x["net_amount"], x["tax_category"]) for x in credit["lines"]] == [(1, "5000.00", "standard")]
with use_cases.repository() as repo:
assert repo.claims.deducted(d1) == D("5000.00") # the credit took what was left: nothing else can deduct it
# ── SO-48 third-party (drop-ship) lines ───────────────────────────────────────────────────────
PRINTER = [{"product_id": "PRINTER-X", "quantity": "2", "unit_code": "C62", "usage": "drop_ship"}]
def test_SO_48_a_drop_ship_line_is_ordered_from_purchasing_once_never_reserved_and_delivered_only_by_the_supplier(neighbours, inventory):
N = neighbours
o = hand(REP, ship_to_bp_id=7313, lines=PRINTER)
assert o["lines"][0]["item_category_code"] == "THIRD_PARTY"
o = confirm(o)
sid = o["sales_order_id"]
req = N.purchasing.requests[f"so:{sid}:1@r1"]
assert (req["product_id"], req["quantity"], req["ship_to_bp_id"], req["sales_order_id"]) == ("PRINTER-X", "2", 7313, sid) and len(N.purchasing.requests) == 1
assert o["lines"][0]["stocked"] is False and o["lines"][0]["purchase_request_id"] == req["request_id"]
assert not [c for c in inventory.calls if c[0] == "reserve" and c[1] == sid] # nothing is reserved in inventory
must(api("POST", f"/sales-orders/{sid}/confirm", REP, {})) # a repeat asks purchasing nothing more
assert len(N.purchasing.requests) == 1
assert "a third-party line is delivered by the supplier" in refused(deliver(o, 1, 2, "DN-1"))
assert api("POST", f"/sales-orders/{sid}/deliveries", FULFILLER, {"line_no": 1, "quantity": "2", "delivery_ref": "pur:SN-55", "source": "purchasing"})[0] == 403
out = must(deliver(o, 1, 2, "pur:SN-55", PURCHASING, source="purchasing"))
assert out["lines"][0]["delivered"] == "2" and out["fulfilment_status"] == "delivered"
again = must(deliver(o, 1, 2, "pur:SN-55", PURCHASING, source="purchasing")) # a repeat records once
assert [d["delivery_ref"] for d in again["deliveries"]] == ["pur:SN-55"]
must(bill(o)) # invoiceable per its policy
assert "supplier_delivery" in refused(deliver(confirm(hand(REP, lines=[{"product_id": "COLA-075", "quantity": "6", "unit_code": "C62"}])), 1, 6, "pur:SN-1",
PURCHASING, source="purchasing"))
def test_SO_48_purchasing_can_refuse_the_cancellation_in_its_own_words(neighbours, inventory):
N = neighbours
o = confirm(hand(REP, ship_to_bp_id=7313, lines=PRINTER))
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"
admin = as_role("order_admin")
said = refused(api("POST", f"/sales-orders/{sid}/cancel", admin, {"reason": "changed our mind"}))
assert "third_party_cancellable" in said and "PO-0144 is confirmed with the supplier" in said and get(sid)["lifecycle"] == "confirmed"
N.purchasing.confirmed_with_supplier.clear() # until purchasing cancels it
assert must(api("POST", f"/sales-orders/{sid}/cancel", admin, {"reason": "changed our mind"}))["lifecycle"] == "cancelled" and N.purchasing.cancelled == [rid]
# ── SO-49 return inspection ───────────────────────────────────────────────────────────────────
def test_SO_49_inspections_record_once_never_above_what_was_received_and_only_inventory_makes_them(neighbours, inventory):
N, o = neighbours, original(neighbours, inventory)
r = confirm(ret(o, "6"))
rid = r["sales_order_id"]
assert "within_received" in refused(inspect(r, 1, 0, ref="inv:RET-1:1")) # nothing is received yet
receive(r, 6)
assert api("POST", f"/sales-orders/{rid}/return-inspections", FULFILLER, {"line_no": 1, "inspection_ref": "x", "received_quantity": "6", "accepted_quantity": "1",
"rejected_quantity": "0", "decision": "restock"})[0] == 403 # so:order:deliver is not so:return:inspect
a = must(inspect(r, 3, 0, ref="inv:RET-1:1", received=6))
assert a["line"]["received"] == "6" and a["line"]["credited"] == "3" and a["lifecycle"] == "confirmed"
must(inspect(r, 3, 0, ref="inv:RET-1:1", received=6)) # repeated: recorded once, credited once
assert len(get(rid)["inspections"]) == 1 and len(N.invoicing.credits) == 1
b = must(inspect(r, 2, 1, ref="inv:RET-1:2", received=6))
assert b["line"]["accepted"] == "5" and b["line"]["rejected"] == "1" and b["line"]["credited"] == "5" and b["lifecycle"] == "done"
assert [x["quantity"] for k in sorted(N.invoicing.credits) for x in N.invoicing.credits[k]["lines"]] == ["3", "2"] # one credit per inspection, never twice
assert "within_received" in refused(inspect(r, 1, 0, ref="inv:RET-1:3", received=6)) # accepted + rejected never above received
std = confirm(hand(REP, lines=COLA24))
assert "only a return is inspected" in refused(inspect(std, 1, 0))