dodil/erp-brick-sales-order
DODIL ERP brick: sales order
tests/test_slice_b.py
"""Slice B of sales-order v6. Rebates: SO-33 (the nightly accrual, each journal once) and SO-34 (partial and final settlement, approved by a
second person, credited once)."""
import json
from datetime import date, datetime
from decimal import Decimal as D
from conftest import ADMIN as ADMIN_ROLE
from conftest import (
BILLER,
INVENTORY,
OWNER,
REP,
SYSTEM,
api,
as_role,
confirm,
deliver,
get,
hand,
must,
on_day,
person,
refused,
)
from erp_platform.repository import CLAIM_TABLES, as_conflict
from sqlalchemy.exc import IntegrityError
from brick_sales_order.application import use_cases
from brick_sales_order.infrastructure import tables
CONTROLLER = as_role("sales_controller")
ADMIN = None # replaced below, once _with_sub exists
def _with_sub(headers, sub):
user = json.loads(headers["X-Dodil-User"])
return {**headers, "X-Dodil-User": json.dumps({**user, "sub": sub})}
ADMIN = _with_sub(ADMIN_ROLE, "admin-sub-1") # the requester: an email AND a different sign-in subject
APPROVER_PERMS = ["so:order:read", "so:rebate:approve"]
SAME_EMAIL = _with_sub(person("order_admin@shop.example", APPROVER_PERMS), "someone-else") # the requester's email, another sub
SAME_SUB = _with_sub(person("second.account@shop.example", APPROVER_PERMS), "admin-sub-1") # another email, the requester's sub
AGREEMENT = {"valid_from": "2027-01-01", "valid_to": "2027-12-31", "basis": "all", "bp_id": 7311, "currency": "GBP", "accrual_rate_pct": "2",
"scale": [{"from_value": "0", "rate_pct": "1"}, {"from_value": "50000", "rate_pct": "2"}, {"from_value": "100000", "rate_pct": "3"}]}
AID = "RB-2027-7311"
def active_agreement(**kw):
with on_day("2026-12-10T09:00:00+00:00"):
must(api("POST", "/rebate-agreements", ADMIN, {**AGREEMENT, **kw}))
must(api("POST", f"/rebate-agreements/{kw.get('rebate_agreement_id', AID)}/activate", ADMIN, {}))
def accrue(day_run, for_date=None):
with on_day(f"{day_run}T09:00:00+00:00"):
return must(api("POST", "/rebate-agreements/accrue", SYSTEM, {"for_date": for_date} if for_date else {}))
def agreement(aid=AID):
return must(api("GET", f"/rebate-agreements/{aid}", ADMIN))
def journals(N):
return list(N.gl.journals.values())
def test_SO_33_the_nightly_accrual_writes_each_line_once_and_each_journal_once(neighbours):
N = neighbours
active_agreement()
N.invoicing.add_billed("INV-2027-0101", 7311, "2027-02-03", [{"net_amount": "10000.00"}])
N.invoicing.add_billed("CN-2027-0007", 7311, "2027-02-10", [{"net_amount": "500.00"}], kind="credit_note")
first = accrue("2027-02-04", "2027-02-03")
second = accrue("2027-02-11", "2027-02-10")
assert first["agreements"][0]["accrued"] == 1 and first["agreements"][0]["amount"] == "200.00" and first["posted"] == 1
assert second["agreements"][0]["amount"] == "-10.00"
calls = len(N.gl.calls)
again = accrue("2027-02-12", "2027-02-03") # the same day again: nothing written, nothing posted
assert again["agreements"][0]["accrued"] == 0 and again["posted"] == 0 and len(N.gl.calls) == calls
a = agreement()
assert a["accruals"] == 2 and a["basis_total"] == "9500.00" and a["accrued_total"] == "190.00"
inv, cn = sorted(journals(N), key=lambda j: j["journal_date"])
assert inv["source_document"] == f"rebate:{AID}:INV-2027-0101" and inv["event"] == "accrual" and inv["journal_date"] == "2027-02-03"
assert inv["lines"] == [{"purpose": "rebate_expense", "debit": "200.00", "bp_id": 7311}, {"purpose": "accrued_rebates", "credit": "200.00", "bp_id": 7311}]
assert cn["lines"] == [{"purpose": "rebate_expense", "credit": "10.00", "bp_id": 7311}, {"purpose": "accrued_rebates", "debit": "10.00", "bp_id": 7311}]
assert inv["journal_id"] != cn["journal_id"] and N.gl.balance("accrued_rebates") == D("190.00")
with use_cases.repository() as repo:
assert [(x.document_id, x.posting_status) for x in repo.rebates.get(AID)[0].accruals] == [("CN-2027-0007", "posted"), ("INV-2027-0101", "posted")]
def test_SO_33_the_basis_is_the_agreements_products_and_the_business_date_is_london_not_utc(neighbours):
N = neighbours
active_agreement(basis="products", basis_values=["COLA-075"])
N.invoicing.add_billed("INV-A", 7311, "2027-07-01", [{"net_amount": "100.00", "product_id": "COLA-075"}, {"net_amount": "900.00", "product_id": "WIDGET"}])
# 23:30 UTC on 1 July is 00:30 BST on 2 July: the previous business day is 1 July, not 30 June
with on_day("2027-07-01T23:30:00+00:00"):
got = must(api("POST", "/rebate-agreements/accrue", SYSTEM, {}))
assert got["for_date"] == "2027-07-01" and got["agreements"][0]["accrued"] == 1 and got["agreements"][0]["amount"] == "2.00"
with on_day("2027-07-01T23:30:00+00:00"):
assert "future" in refused(api("POST", "/rebate-agreements/accrue", SYSTEM, {"for_date": "2027-07-03"}), 422)
assert api("POST", "/rebate-agreements/accrue", ADMIN, {})[0] == 403
def test_SO_33_a_journal_the_ledger_refuses_stays_pending_with_its_reason_and_is_retried(neighbours):
N = neighbours
active_agreement()
N.invoicing.add_billed("INV-2027-0300", 7311, "2027-03-05", [{"net_amount": "1000.00"}])
N.gl.closed_before = date(2027, 4, 1)
first = accrue("2027-03-06", "2027-03-05")
assert first["agreements"][0]["accrued"] == 1 and first["pending"] == 1 and not N.gl.journals
with use_cases.repository() as repo:
post = repo.postings.get(f"rebate:{AID}:INV-2027-0300", "accrual")
assert post["status"] == "pending" and "period closed" in post["reason"] and post["attempts"] == 1
N.gl.closed_before = None
retried = accrue("2027-03-07", "2027-03-06")
assert retried["posted"] == 1 and retried["pending"] == 0 and len(N.gl.journals) == 1
assert accrue("2027-03-08", "2027-03-06")["posted"] == 0 and len(N.gl.journals) == 1
assert agreement()["accruals"] == 1
def test_SO_33_without_the_ledger_the_accrual_is_reported_not_posted(neighbours):
N = neighbours
active_agreement()
with on_day("2026-12-11T09:00:00+00:00"):
must(api("POST", "/controls/rebate_accrual_mode", OWNER, {"value": "report", "valid_from": "2026-12-12"}))
N.invoicing.add_billed("INV-R", 7311, "2027-03-05", [{"net_amount": "1000.00"}])
got = accrue("2027-03-06", "2027-03-05")
assert got["agreements"][0]["accrued"] == 1 and not N.gl.calls
with use_cases.repository() as repo:
assert repo.postings.get(f"rebate:{AID}:INV-R", "accrual")["status"] == "reported"
assert "post, report" in refused(api("POST", "/controls/rebate_accrual_mode", OWNER, {"value": "monthly", "valid_from": "2026-12-13"}), 422)
def test_SO_33_the_agreements_steps_share_one_claim_key_and_a_racer_is_a_409(neighbours):
assert "so_rebate_claim" in CLAIM_TABLES
active_agreement()
with use_cases.repository() as repo:
assert repo.rebates.claim(AID, "accrue") == 1 and repo.rebates.claim(AID, "change_scale") == 2
racer = None
try: # a second writer that looked at step 2 too
repo.conn.execute(tables.so_rebate_claim.insert().values(rebate_agreement_id=AID, seq=2, operation="accrue"))
except IntegrityError as e:
racer = as_conflict(e)
assert racer is not None
assert api("POST", "/rebate-agreements", ADMIN, {**AGREEMENT, "accrual_rate_pct": "101", "rebate_agreement_id": "RB-X"})[0] == 422
def _year(N):
"""The 2027 of the card: 50,000.00 of drinks by the end of June, 70,000.00 more later: 120,000.00 in all, two tax categories."""
N.invoicing.add_billed("INV-2027-0101", 7311, "2027-02-03", [{"net_amount": "10000.00"}])
N.invoicing.add_billed("CN-2027-0007", 7311, "2027-02-10", [{"net_amount": "500.00"}], kind="credit_note")
N.invoicing.add_billed("INV-2027-0150", 7311, "2027-05-04", [{"net_amount": "40500.00"}])
N.invoicing.add_billed("INV-2027-0300", 7311, "2027-09-06", [{"net_amount": "40000.00", "tax_category": "reduced"}, {"net_amount": "30000.00"}])
for d in ("2027-02-03", "2027-02-10", "2027-05-04"):
accrue(f"{d[:8]}{int(d[8:]) + 1:02d}", d)
def test_SO_34_a_partial_then_the_final_settlement_two_people_each_credited_once(neighbours):
N = neighbours
active_agreement()
_year(N)
assert agreement()["basis_total"] == "50000.00"
with on_day("2027-07-05T09:00:00+00:00"): # 2% of 50,000.00 = 1,000.00 reached so far
asked = must(api("POST", f"/rebate-agreements/{AID}/partial-settlements", ADMIN, {"reason": "mid-year payment"}))
assert asked["settlement_no"] == 1 and asked["settlements"][0]["amount"] == "1000.00"
assert "waiting for approval" in refused(api("POST", f"/rebate-agreements/{AID}/partial-settlements", ADMIN, {"reason": "again"}))
for second in (SAME_EMAIL, SAME_SUB): # the requester under another account: same email, or same sign-in subject
assert "approver_not_requester" in refused(api("POST", f"/rebate-agreements/{AID}/partial-settlements/1/approve", second, {}))
assert not N.invoicing.credits
done = must(api("POST", f"/rebate-agreements/{AID}/partial-settlements/1/approve", CONTROLLER, {}))
assert done["settlements"][0]["status"] == "credited" and done["settled_total"] == "1000.00"
must(api("POST", f"/rebate-agreements/{AID}/partial-settlements/1/approve", CONTROLLER, {})) # approving twice credits once
assert list(N.invoicing.credits) == [f"rebate:{AID}:1"]
assert N.invoicing.credits[f"rebate:{AID}:1"]["account_purpose"] == "rebate"
for d in ("2027-09-06",):
accrue("2027-09-07", d)
assert agreement()["basis_total"] == "120000.00" and agreement()["accrued_total"] == "2400.00"
assert api("POST", f"/rebate-agreements/{AID}/final-settlement", ADMIN, {"reason": "year end"})[0] == 409 # not ended yet
with on_day("2028-01-02T09:00:00+00:00"): # the nightly run after the last day ends the agreement
ran = must(api("POST", "/rebate-agreements/accrue", SYSTEM, {"for_date": "2027-12-31"}))
assert ran["ended"] == [AID]
assert agreement()["lifecycle"] == "ended"
with on_day("2028-01-05T09:00:00+00:00"):
asked = must(api("POST", f"/rebate-agreements/{AID}/final-settlement", ADMIN, {"reason": "year end"}))
fin = asked["settlements"][1]
assert (fin["basis_amount"], fin["rate_pct"], fin["earned_amount"], fin["settled_before"], fin["amount"], fin["true_up_amount"]) == \
("120000.00", "3", "3600.00", "1000.00", "2600.00", "1200.00")
assert sum(D(x["net_amount"]) for x in fin["lines"]) == D("2600.00") and {x["tax_category"] for x in fin["lines"]} == {"standard", "reduced"}
for second in (SAME_EMAIL, SAME_SUB):
assert "approver_not_requester" in refused(api("POST", f"/rebate-agreements/{AID}/final-settlement/2/approve", second, {}))
before = len(N.gl.calls)
done = must(api("POST", f"/rebate-agreements/{AID}/final-settlement/2/approve", CONTROLLER, {}))
again = must(api("POST", f"/rebate-agreements/{AID}/final-settlement/2/approve", CONTROLLER, {}))
assert done == again and done["lifecycle"] == "settled" and done["accrual_balance"] == "0.00"
assert len(N.gl.calls) == before + 1 # the true-up, once
true_up = next(j for j in journals(N) if j["event"] == "true_up")
assert true_up["source_document"] == f"rebate:{AID}:2" and true_up["lines"][0]["debit"] == "1200.00"
credit = N.invoicing.credits[f"rebate:{AID}:2"]
assert credit["account_purpose"] == "rebate" and sum(D(x["net_amount"]) for x in credit["lines"]) == D("2600.00")
assert sorted(N.invoicing.credits) == [f"rebate:{AID}:1", f"rebate:{AID}:2"]
# accrued 2,400 + true-up 1,200 less credited 1,000 and 2,600 nets to exactly zero in the ledger's accrued-rebates account
assert N.gl.balance("accrued_rebates") == D("3600.00") and D("3600.00") - D("1000.00") - D("2600.00") == 0
def test_SO_34_the_ledger_and_invoicing_are_asked_only_after_the_claim_and_a_refusal_credits_nothing(neighbours):
N = neighbours
active_agreement()
_year(N)
accrue("2027-09-07", "2027-09-06")
accrue("2028-01-02", "2027-12-31")
with on_day("2028-01-05T09:00:00+00:00"):
must(api("POST", f"/rebate-agreements/{AID}/final-settlement", ADMIN, {"reason": "year end"}))
N.gl.closed_before = date(2028, 2, 1) # the period of the true-up is closed
said = refused(api("POST", f"/rebate-agreements/{AID}/final-settlement/1/approve", CONTROLLER, {}))
assert "ledger" in said and "nothing was credited" in said and not N.invoicing.credits
a = agreement()
assert a["settlements"][0]["status"] == "approved" and a["lifecycle"] == "ended"
N.gl.closed_before = None
N.invoicing.down = True
assert api("POST", f"/rebate-agreements/{AID}/final-settlement/1/approve", CONTROLLER, {})[0] == 502
N.invoicing.down = False
done = must(api("POST", f"/rebate-agreements/{AID}/final-settlement/1/approve", CONTROLLER, {}))
assert done["lifecycle"] == "settled" and len(N.invoicing.credits) == 1 and done["accrual_balance"] == "0.00"
assert sum(1 for j in journals(N) if j["event"] == "true_up") == 1
def test_SO_34_nothing_to_settle_is_refused_and_a_settlement_that_does_not_exist_is_a_409(neighbours):
active_agreement()
with on_day("2027-02-05T09:00:00+00:00"):
assert "something_to_settle" in refused(api("POST", f"/rebate-agreements/{AID}/partial-settlements", ADMIN, {"reason": "r"}))
assert api("POST", f"/rebate-agreements/{AID}/partial-settlements/1/approve", CONTROLLER, {})[0] == 409
# ═══ SO-37 to SO-40: partners, incompletion, blocks, delivery rules ═══════════════════════════════
WID = [{"product_id": "WIDGET", "quantity": "100", "unit_code": "C62"}]
EARLIER = "2026-09-20T09:00:00+00:00"
def partner_rows(sold_to=7311):
with on_day(EARLIER):
for fn, bp, default, transit in (("ship_to", 7312, True, [{"warehouse_id": "WH1", "days": 1}]), ("ship_to", 7313, False, []),
("bill_to", 7311, True, []), ("payer", 7400, True, [])):
must(api("POST", "/customer-partners", ADMIN_ROLE, {"sold_to_bp_id": sold_to, "function": fn, "partner_bp_id": bp, "valid_from": "2026-10-01",
"is_default": default, "transit_days": transit}))
def test_SO_40_partners_default_from_the_rows_in_force_are_copied_and_each_function_is_checked(neighbours):
N = neighbours
N.partners.customers[7312] = True
partner_rows()
o = hand(REP, lines=WID)
assert (o["ship_to_bp_id"], o["bill_to_bp_id"], o["payer_bp_id"]) == ("7312", "7311", "7400")
picked = hand(REP, lines=WID, ship_to_bp_id=7313, ship_to_country="DE")
assert picked["ship_to_bp_id"] == "7313"
assert "9999 is not a ship-to of 7311" in refused(api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "lines": WID, "ship_to_bp_id": 9999}), 422)
assert "starts after today" in refused(api("POST", "/customer-partners", ADMIN_ROLE, {"sold_to_bp_id": 7311, "function": "payer", "partner_bp_id": 7400,
"valid_from": "2026-11-20"}), 422)
N.payments.status[7400] = {"hold": True, "reason": "Group Treasury is over its limit", "mode": "block"} # the PAYER's hold, not the sold-to's
said = refused(api("POST", f"/sales-orders/{picked['sales_order_id']}/confirm", REP, {}))
assert "credit" in said and "Group Treasury" in said
N.payments.status.clear()
assert N.tax.countries[-1] == "DE" # tax for the ship-to's country
assert confirm(picked)["lifecycle"] == "confirmed"
c = confirm(hand(REP, lines=WID, ship_to_bp_id=7313, ship_to_country="GB"))
deliver(c, 1, 100, "D1")
must(api("POST", f"/sales-orders/{c['sales_order_id']}/request-invoice", BILLER, {}))
d = next(iter(N.invoicing.drafts.values()))
assert (d["bp_id"], d["sold_to_bp_id"], d["ship_to_bp_id"], d["bill_to_bp_id"]) == (7400, 7311, 7313, 7311)
def test_SO_40_v7_the_ship_tos_country_is_copied_from_business_partner_when_the_order_names_none(neighbours):
"""v7 (the business-partner seam): GET /partners/{bp_id} answers country; the order copies the ship-to's as ship_to_country, which
tax's place of supply and the export incompletion rule read. Before, an order took a country only when one was typed."""
N = neighbours
N.partners.customers[7312] = True
N.partners.countries[7313] = "DE"
partner_rows()
assert hand(REP, lines=WID)["ship_to_country"] == "GB" # the default ship-to 7312 is in GB
abroad = hand(REP, lines=WID, ship_to_bp_id=7313)
assert abroad["ship_to_country"] == "DE" and N.tax.countries[-1] == "DE"
assert hand(REP, lines=WID, ship_to_bp_id=7313, ship_to_country="FR")["ship_to_country"] == "FR" # typed wins
def test_SO_40_a_partner_must_be_an_active_customer_and_does_not_change_after_the_first_delivery(neighbours):
N = neighbours
N.partners.customers[7312] = True
partner_rows()
with on_day(EARLIER):
must(api("POST", "/customer-partners", ADMIN_ROLE, {"sold_to_bp_id": 7311, "function": "payer", "partner_bp_id": 7500, "valid_from": "2026-10-01"}))
supplier = hand(REP, lines=WID, payer_bp_id=7500)
assert "partners_valid" in refused(api("POST", f"/sales-orders/{supplier['sales_order_id']}/confirm", REP, {})) and "payer 7500" in \
refused(api("POST", f"/sales-orders/{supplier['sales_order_id']}/confirm", REP, {}))
o = confirm(hand(REP, lines=WID))
moved = must(api("POST", f"/sales-orders/{o['sales_order_id']}/partners", REP, {"ship_to_bp_id": 7313}))
assert moved["ship_to_bp_id"] == "7313"
assert api("POST", f"/sales-orders/{o['sales_order_id']}/partners", REP, {"ship_to_bp_id": 9999})[0] == 422
deliver(o, 1, 10, "D1")
assert "partners_unchanged_after_delivery" in refused(api("POST", f"/sales-orders/{o['sales_order_id']}/partners", REP, {"payer_bp_id": 7311}))
def test_SO_38_an_export_order_confirms_but_its_delivery_and_billing_wait_for_the_fields_the_rules_ask(neighbours):
o = confirm(hand(REP, lines=WID, ship_to_country="DE")) # no delivery terms: confirmation is allowed
ready = must(api("GET", f"/sales-orders/{o['sales_order_id']}/readiness", REP))["readiness"]
miss = next(g for g in ready["delivery"] if g["gate"] == "incomplete:delivery_terms")
assert miss == {"gate": "incomplete:delivery_terms", "met": False, "reason": "an export order needs its delivery terms (Incoterms)", "fix": "Order: delivery terms"}
fulfil = must(api("GET", f"/sales-orders/{o['sales_order_id']}/fulfilment", INVENTORY))
assert fulfil["delivery_blocked"]["met"] is False and fulfil["delivery_blocked"]["reasons"][0]["gate"] == "incomplete:delivery_terms"
assert "incomplete:delivery_terms" in refused(deliver(o, 1, 5, "D1"))
home = confirm(hand(REP, lines=WID, ship_to_country="GB")) # a domestic order has no such rule
assert all(g["met"] for g in must(api("GET", f"/sales-orders/{home['sales_order_id']}/readiness", REP))["readiness"]["delivery"])
def test_SO_38_a_customer_that_asks_for_a_po_blocks_billing_and_a_return_needs_its_reason(neighbours):
from brick_sales_order.application import use_cases as uc
with on_day(EARLIER), uc.repository() as repo:
repo.policies.put({"bp_id": 7311, "valid_from": date(2026, 10, 1), "po_required": True, "deposit_pct": None, "delivery_terms": "", "set_by": "t",
"price_group": ""})
o = confirm(hand(REP, lines=WID, customer_po_ref="PO-1"))
bare = must(api("GET", f"/sales-orders/{o['sales_order_id']}/readiness", REP))["readiness"]
assert next(g for g in bare["billing"] if g["gate"] == "incomplete:customer_po_ref")["met"] is True
must(deliver(o, 1, 100, "DN-1"))
ret = {"order_type_code": "RETURNS", "ref_sales_order_id": o["sales_order_id"], "ref_line_no": 1, "customer_po_ref": "PO-2", "ref_delivery_ref": "DN-1",
"ref_invoice_id": "INV-1", "ref_invoice_line_no": 1}
said = refused(api("POST", f"/sales-orders/{hand(REP, lines=[{'product_id': 'WIDGET', 'quantity': '10'}], **ret)['sales_order_id']}/confirm", REP, {}))
assert "incomplete:return_reason" in said and "a return needs its reason" in said
assert confirm(hand(REP, lines=[{"product_id": "WIDGET", "quantity": "10"}], return_reason="damaged", **ret))["lifecycle"] == "confirmed"
with on_day(EARLIER):
must(api("POST", "/incompletion-rules", ADMIN_ROLE, {"order_type_code": "STANDARD", "field": "customer_po_ref", "valid_from": "2026-10-01",
"blocks": "billing", "condition": "always", "reason_text": "every invoice names a PO"}))
nopo = confirm(hand(REP, bp_id=8420, lines=WID))
deliver(nopo, 1, 100, "D1")
assert "incomplete:customer_po_ref" in refused(api("POST", f"/sales-orders/{nopo['sales_order_id']}/request-invoice", BILLER, {}))
assert api("POST", "/incompletion-rules", REP, {"order_type_code": "STANDARD", "field": "customer_po_ref", "valid_from": "2026-12-01", "blocks": "billing",
"condition": "always", "reason_text": "r"})[0] == 403
def test_SO_39_a_block_is_an_appended_row_and_stops_delivery_or_billing_until_its_release(neighbours):
for code, kind in (("CHECK_TERMS", "delivery"), ("PRICE_DISPUTE", "billing")):
must(api("POST", "/block-reasons", ADMIN_ROLE, {"block_reason_code": code, "kind": kind, "name": code}))
o = confirm(hand(REP, lines=WID))
oid = o["sales_order_id"]
assert api("POST", f"/sales-orders/{oid}/blocks", REP, {"kind": "delivery", "reason_code": "CHECK_TERMS", "note": "n"})[0] == 403
assert api("POST", f"/sales-orders/{oid}/blocks", ADMIN_ROLE, {"kind": "delivery", "reason_code": "PRICE_DISPUTE", "note": "n"})[0] == 422
must(api("POST", f"/sales-orders/{oid}/blocks", ADMIN_ROLE, {"kind": "delivery", "reason_code": "CHECK_TERMS", "note": "terms to check"}))
held = must(api("GET", f"/sales-orders/{oid}/fulfilment", INVENTORY))["delivery_blocked"]
assert held["met"] is False and held["reasons"][0]["gate"] == "delivery_block" and "CHECK_TERMS" in held["reasons"][0]["reason"]
assert "delivery_block" in refused(deliver(o, 1, 10, "D1"))
assert api("POST", f"/sales-orders/{oid}/blocks/release", REP, {"kind": "delivery", "note": "ok"})[0] == 403 # the rep may not release
must(api("POST", f"/sales-orders/{oid}/blocks/release", ADMIN_ROLE, {"kind": "delivery", "note": "terms checked"}))
assert must(api("GET", f"/sales-orders/{oid}/fulfilment", INVENTORY))["delivery_blocked"]["met"] is True
assert deliver(o, 1, 10, "D1")[0] == 200
rows = get(oid)["blocks"]
assert [(b["block_no"], b["event"], b["reason_code"], b["by_email"]) for b in rows] == [(1, "set", "CHECK_TERMS", "order_admin@shop.example"),
(2, "release", "CHECK_TERMS", "order_admin@shop.example")]
must(api("POST", f"/sales-orders/{oid}/blocks", ADMIN_ROLE, {"kind": "billing", "reason_code": "PRICE_DISPUTE", "note": "customer disputes the price"}))
assert "on hold: PRICE_DISPUTE" in refused(api("POST", f"/sales-orders/{oid}/request-invoice", BILLER, {}))
assert not neighbours.invoicing.drafts
def test_SO_39_a_free_of_charge_order_starts_with_the_approval_block_which_no_one_releases_by_hand(neighbours):
o = confirm(hand(REP, order_type_code="FREE_OF_CHARGE", lines=WID))
assert [(b["kind"], b["reason_code"], b["event"]) for b in get(o["sales_order_id"])["blocks"]] == [("delivery", "APPROVAL", "set"), ("billing", "APPROVAL", "set")]
said = refused(api("POST", f"/sales-orders/{o['sales_order_id']}/blocks/release", ADMIN_ROLE, {"kind": "delivery", "note": "n"}))
assert "approval_block" in said
assert "delivery_block" in refused(deliver(o, 1, 1, "D1"))
def test_SO_37_the_customers_delivery_rules_are_copied_at_confirmation_and_enforced_per_line(neighbours):
from brick_sales_order.application import use_cases as uc
with on_day(EARLIER), uc.repository() as repo:
base = {"deposit_pct": None, "delivery_terms": "", "set_by": "t", "price_group": "", "po_required": False}
repo.policies.put({**base, "bp_id": 7311, "valid_from": date(2026, 10, 1), "partial_delivery": "allowed", "max_partials": 3,
"under_tolerance_pct": D("2"), "over_tolerance_pct": D("0"), "combine_allowed": True, "min_shelf_life_days": 30})
repo.policies.put({**base, "bp_id": 8420, "valid_from": date(2026, 10, 1), "partial_delivery": "complete_order", "max_partials": 0,
"under_tolerance_pct": D("0"), "over_tolerance_pct": D("0"), "combine_allowed": False, "min_shelf_life_days": 0})
three = WID + [{"product_id": "WIDGET", "quantity": "100", "unit_code": "C62"}, {"product_id": "WIDGET", "quantity": "100", "unit_code": "C62"}]
acme = confirm(hand(REP, lines=three))
other = confirm(hand(REP, bp_id=8420, lines=WID))
a, b = (must(api("GET", f"/sales-orders/{x['sales_order_id']}/fulfilment", INVENTORY)) for x in (acme, other))
assert (a["partial_delivery"], a["max_partials"], a["under_tolerance_pct"], a["over_tolerance_pct"], a["combine_allowed"], a["min_remaining_shelf_life_days"]) == \
("allowed", 3, "2", "0", True, 30)
assert b["partial_delivery"] == "complete_order" and not any("price" in k or "net" in k for ln in a["lines"] for k in ln) and a["purpose"] == "sale"
assert deliver(acme, 1, 98, "D1")[0] == 200 # 98 of 100 is within the 2%: the line is closed
for n, ref in enumerate(("P1", "P2", "P3"), 1):
assert deliver(acme, 2, 30, ref)[0] == 200
assert "max_partials" in refused(deliver(acme, 2, 10, "P4")) # a fourth part delivery
assert "within_tolerance" in refused(deliver(acme, 3, 101, "X1")) # 101 on a line of 100, over-delivery 0%
fx = must(api("GET", f"/sales-orders/{acme['sales_order_id']}/fulfilment", INVENTORY))
assert next(x for x in fx["lines"] if x["line_no"] == 1)["closed"] is True and next(x for x in fx["lines"] if x["line_no"] == 2)["closed"] is False
must(api("POST", f"/sales-orders/{acme['sales_order_id']}/request-invoice", BILLER, {}))
d = next(iter(N_drafts(neighbours)))
assert next(x for x in d["lines"] if x["line_no"] == 1)["quantity"] == "98" # invoicing bills what was delivered: 98
assert api("POST", f"/sales-orders/{acme['sales_order_id']}/fulfilment-rules", REP, {"max_partials": 5})[0] == 409 # fixed from the first delivery
fresh = confirm(hand(REP, lines=WID))
assert must(api("POST", f"/sales-orders/{fresh['sales_order_id']}/fulfilment-rules", REP, {"max_partials": 5, "over_tolerance_pct": "1"}))["max_partials"] == 5
assert deliver(fresh, 1, 101, "Y1")[0] == 200 # the rep allowed 1% over: 101 of 100 is within it
def N_drafts(N):
return N.invoicing.drafts.values()
# ═══ SO-35 and SO-36: promised dates, and the daily re-promise ═════════════════════════════════════════
DAY1, DAY2 = "2026-11-02T09:00:00+00:00", "2026-11-05T09:00:00+00:00"
JUICE = [{"product_id": "JUICE-075", "quantity": "480", "unit_code": "C62"}] # 1 C62 = 0.75 LTR: 480 C62 = 360 LTR
def warehouse_set(wh="WH1"):
from brick_sales_order.application import use_cases as uc
with on_day(EARLIER), uc.repository() as repo:
repo.settings.put("default_warehouse_id", date(2026, 10, 1), wh, "t", datetime(2026, 9, 20))
def setting(key, value, valid_from=date(2026, 10, 1)):
from brick_sales_order.application import use_cases as uc
with on_day(EARLIER), uc.repository() as repo:
repo.settings.put(key, valid_from, value, "t", datetime(2026, 9, 20))
def story_stock(inventory):
"""360 LTR on hand less 180 reserved by others = 180 now; PO-0101 brings 720 LTR on 2026-11-10 (goods-in 1 day: inventory's date)."""
inventory.supply["JUICE-075"] = [(date(2026, 11, 2), D("180"), None), (date(2026, 11, 10), D("720"), None)]
def rows(order_id, line=1):
return [(r["schedule_version"], r["confirmed_on"], r["requested_quantity"]) for r in get(order_id)["lines"][line - 1]["schedule"]]
def test_SO_35_a_shortfall_gives_two_schedule_lines_shown_and_printed_and_never_blocks_confirmation(neighbours, inventory):
neighbours.partners.customers[7312] = True
partner_rows()
warehouse_set()
story_stock(inventory)
with on_day(DAY1):
o = hand(REP, lines=JUICE, requested_delivery_on="2026-11-04")
drafted = must(api("POST", f"/sales-orders/{o['sales_order_id']}/promise-dates", REP, {}))
assert [(r["confirmed_on"], r["requested_quantity"], r["confirmed_quantity"]) for r in drafted["lines"][0]["schedule"]] == \
[("2026-11-04", "240", "240"), ("2026-11-12", "240", "240")]
assert drafted["lines"][0]["promise"] == "partly confirmed: 240 on 4 Nov, 240 on 12 Nov"
q = inventory.atp_calls[0] # the line's own site, quantity, unit and wanted date
assert (q["warehouse_id"], q["quantity"], q["unit_code"], q["requested_on"]) == ("WH1", "480", "C62", "2026-11-02") # 4 Nov less pick-pack 1 and transit 1
c = confirm(o) # a shortfall never blocks
assert c["lifecycle"] == "confirmed" and c["schedule_version"] == 1 # confirm re-promised the same answer: no second version
sent = must(api("POST", f"/sales-orders/{o['sales_order_id']}/send-confirmation", REP, {"address": "buyer@acme.example"}))
assert sent["document"]["lines"][0]["promise"] == "partly confirmed: 240 on 4 Nov, 240 on 12 Nov" and len(sent["document"]["lines"][0]["schedule"]) == 2
who, body = inventory.reserve_bodies[-1] # the reservation carries the promise, same revision
assert body["revision"] == 1 and body["schedule_version"] == 1 and body["priority"] == "normal"
assert body["lines"][0]["warehouse_id"] == "WH1" and body["lines"][0]["schedule"] == \
[{"confirmed_on": "2026-11-04", "quantity_base": "180"}, {"confirmed_on": "2026-11-12", "quantity_base": "180"}]
assert api("POST", f"/sales-orders/{c['sales_order_id']}/promise-dates", REP, {})[0] == 409 # dates are promised on a draft
def test_SO_35_a_lot_short_of_the_minimum_shelf_life_is_not_counted_and_complete_delivery_promises_the_whole_line_late(neighbours, inventory):
from brick_sales_order.application import use_cases as uc
neighbours.partners.customers[7312] = True
partner_rows()
partner_rows(8420)
warehouse_set()
inventory.supply["JUICE-075"] = [(date(2026, 11, 2), D("180"), date(2026, 11, 20)), (date(2026, 11, 10), D("720"), None)] # the first lot expires 20 Nov
with on_day(EARLIER), uc.repository() as repo:
base = {"deposit_pct": None, "delivery_terms": "", "set_by": "t", "price_group": "", "po_required": False, "max_partials": 0,
"under_tolerance_pct": D("0"), "over_tolerance_pct": D("0"), "combine_allowed": False}
repo.policies.put({**base, "bp_id": 7311, "valid_from": date(2026, 10, 1), "partial_delivery": "allowed", "min_shelf_life_days": 30})
repo.policies.put({**base, "bp_id": 8420, "valid_from": date(2026, 10, 1), "partial_delivery": "complete_order", "min_shelf_life_days": 0})
with on_day(DAY1):
short = confirm(hand(REP, lines=JUICE, requested_delivery_on="2026-11-04"))
assert inventory.atp_calls[-1]["min_shelf_life_days"] == 30 # the customer's minimum, from the policy copied at confirmation
assert rows(short["sales_order_id"]) == [(1, "2026-11-12", "480")] # the 180 LTR lot (30 days from 2 Nov > 20 Nov) is not counted
inventory.supply["JUICE-075"] = [(date(2026, 11, 2), D("180"), None), (date(2026, 11, 10), D("720"), None)]
whole = confirm(hand(REP, bp_id=8420, lines=JUICE, requested_delivery_on="2026-11-04"))
assert [(r["confirmed_on"], r["requested_quantity"]) for r in get(whole["sales_order_id"])["lines"][0]["schedule"]] == [("2026-11-12", "480")]
assert get(whole["sales_order_id"])["lines"][0]["promise"] == "confirmed: 480 on 12 Nov"
def test_SO_35_without_inventory_or_a_warehouse_or_stock_the_requested_date_is_kept_and_nothing_was_checked(neighbours, inventory):
with on_day(DAY1):
none = confirm(hand(REP, lines=JUICE, requested_delivery_on="2026-11-04")) # inventory installed, no delivering warehouse set
assert rows(none["sales_order_id"]) == [(1, "2026-11-04", "480")] and not inventory.atp_calls
assert "availability not checked: no delivering warehouse is set" in get(none["sales_order_id"])["lines"][0]["promise"]
warehouse_set()
mixed = confirm(hand(REP, lines=[*JUICE, {"product_id": "SETUP", "quantity": "3", "unit_code": "HUR"}], requested_delivery_on="2026-11-04"))
service = get(mixed["sales_order_id"])["lines"][1]
assert [(r["confirmed_on"], r["reason"]) for r in service["schedule"]] == [("2026-11-04", "availability not checked: not a stocked line")]
inventory.atp_down = True # an inventory that does not answer never blocks either
down = confirm(hand(REP, lines=JUICE, requested_delivery_on="2026-11-04"))
assert "availability not checked" in get(down["sales_order_id"])["lines"][0]["promise"] and down["lifecycle"] == "confirmed"
def test_SO_35_no_inventory_in_the_installation_keeps_the_requested_date_and_says_so(neighbours):
with on_day(DAY1):
o = confirm(hand(REP, lines=JUICE, requested_delivery_on="2026-11-04"))
assert rows(o["sales_order_id"]) == [(1, "2026-11-04", "480")]
assert get(o["sales_order_id"])["lines"][0]["promise"].startswith("availability not checked: inventory is not installed")
def test_SO_35_what_inventory_cannot_cover_is_an_unconfirmed_quantity_and_the_schedule_never_over_confirms(neighbours, inventory):
neighbours.partners.customers[7312] = True
partner_rows()
warehouse_set()
inventory.supply["JUICE-075"] = [(date(2026, 11, 2), D("75"), None)] # 75 LTR = 100 C62, nothing more is due
with on_day(DAY1):
o = confirm(hand(REP, lines=JUICE, requested_delivery_on="2026-11-04"))
assert o["lifecycle"] == "confirmed" and get(o["sales_order_id"])["lines"][0]["promise"] == "partly confirmed: 100 on 4 Nov, 380 not yet confirmed"
from brick_sales_order.application import use_cases as uc
from brick_sales_order.domain.errors import RuleBroken
with uc.repository() as repo:
agg, _ = repo.sales_orders.get(o["sales_order_id"])
import pytest
with pytest.raises(RuleBroken, match="versions only increase"):
agg.append_schedule(1, 1, [], requested_on=None, reason="x", at=datetime(2026, 11, 2))
with pytest.raises(RuleBroken, match="confirm 481 of 480"):
from brick_sales_order.domain.promise import Chunk
agg.append_schedule(2, 1, [Chunk(date(2026, 11, 5), D("481"))], requested_on=None, reason="x", at=datetime(2026, 11, 2))
def promised_240(inventory, neighbours, **hand_kw):
"""The SO-36 start: line 1 is 240 C62 confirmed for 2026-11-12 against the purchase order arriving (inventory: 2026-11-10)."""
neighbours.partners.customers[7312] = True
partner_rows()
warehouse_set()
inventory.supply["JUICE-075"] = [(date(2026, 11, 10), D("720"), None)]
with on_day(DAY1):
o = confirm(hand(REP, lines=[{"product_id": "JUICE-075", "quantity": "240", "unit_code": "C62"}], requested_delivery_on="2026-11-04", **hand_kw))
assert rows(o["sales_order_id"]) == [(1, "2026-11-12", "240")]
return o
def purchase_moved(inventory):
"""PO-0101 moves to 16 Nov (goods in on the 17th) and another order's cancellation frees 90 LTR today (5 Nov)."""
inventory.supply["JUICE-075"] = [(date(2026, 11, 5), D("90"), None), (date(2026, 11, 17), D("720"), None)]
def reschedule(who=SYSTEM, **body):
return must(api("POST", "/sales-orders/reschedule", who, body))
def test_SO_36_the_re_promise_appends_a_version_tells_the_owner_resends_the_reservation_and_a_second_run_writes_nothing(neighbours, inventory):
o = promised_240(inventory, neighbours)
oid = o["sales_order_id"]
purchase_moved(inventory)
with on_day(DAY2):
first = reschedule()
assert first["business_date"] == "2026-11-05" and first["re_promised"] == [oid] and first["tasks"] == 1 and first["reservation_pending"] == 0
assert rows(oid) == [(2, "2026-11-07", "120"), (2, "2026-11-19", "120")] # 120 from the freed 90 LTR, 120 from the moved purchase
reasons = {r["reason"] for r in get(oid)["lines"][0]["schedule"]}
assert reasons == {"re-promised 2026-11-05 (daily)"}
tasks = must(api("GET", "/tasks", REP))["tasks"]
assert [(t["sales_order_id"], t["kind"], t["text"]) for t in tasks] == [(oid, "repromise", "tell 7311: 120 C62 now 19 Nov")]
_, body = inventory.reserve_bodies[-1] # the SAME revision, the new schedule version
assert (body["revision"], body["schedule_version"]) == (1, 2)
assert body["lines"][0]["schedule"] == [{"confirmed_on": "2026-11-07", "quantity_base": "90"}, {"confirmed_on": "2026-11-19", "quantity_base": "90"}]
sent_before, written = len(inventory.reserve_bodies), get(oid)["version"]
inventory.supply["JUICE-075"] = [(date(2026, 11, 5), D("300"), None), (date(2026, 11, 17), D("720"), None)] # stock changes during the day
again = reschedule()
assert again["re_promised"] == [] and again["tasks"] == 0 # the second run the same day writes nothing: once per business date
assert get(oid)["version"] == written and len(inventory.reserve_bodies) == sent_before
assert len(must(api("GET", "/tasks", REP))["tasks"]) == 1
assert api("POST", "/sales-orders/reschedule", REP, {})[0] == 403 # a machine's route: no person's role reaches it
assert get(oid)["schedule_version"] == 2
def test_SO_36_a_date_the_customer_was_told_is_never_pushed_earlier_on_its_own_it_is_offered(neighbours, inventory):
o = promised_240(inventory, neighbours)
oid = o["sales_order_id"]
with on_day(DAY1):
must(api("POST", f"/sales-orders/{oid}/send-confirmation", REP, {"address": "buyer@acme.example"})) # version 1 is now the customer's
purchase_moved(inventory)
with on_day(DAY2):
reschedule()
assert rows(oid) == [(2, "2026-11-12", "120"), (2, "2026-11-19", "120")] # the told 12 Nov stays for 120; the later one moves
texts = {t["kind"]: t["text"] for t in must(api("GET", "/tasks", REP))["tasks"]}
assert texts == {"repromise": "tell 7311: 120 C62 now 19 Nov", "offer": "offer 7311: 120 C62 could go on 7 Nov"}
setting("reschedule_mode", "both")
inventory.supply["JUICE-075"] = [(date(2026, 11, 6), D("180"), None), (date(2026, 11, 17), D("720"), None)]
with on_day("2026-11-06T09:00:00+00:00"):
reschedule() # mode both: the earlier date is applied on its own
assert [(v, on) for v, on, _ in rows(oid)][0] == (3, "2026-11-08") # 180 LTR today = 240 C62: all of it, applied on the 8th (6 Nov + 2)
def test_SO_36_rush_orders_are_re_promised_first_then_by_order_date_and_a_cancellation_triggers_it(neighbours, inventory):
neighbours.partners.customers[7312] = True
partner_rows()
warehouse_set()
inventory.supply["JUICE-075"] = [(date(2026, 11, 10), D("720"), None)]
line = [{"product_id": "JUICE-075", "quantity": "240", "unit_code": "C62"}]
with on_day(DAY1):
first = confirm(hand(REP, lines=line, requested_delivery_on="2026-11-04", sales_order_id="A-normal")) # sorts before the rush order by id
rush = confirm(hand(REP, lines=line, requested_delivery_on="2026-11-04", priority="rush", sales_order_id="Z-rush"))
victim = confirm(hand(REP, lines=[{"product_id": "JUICE-075", "quantity": "12", "unit_code": "C62"}]))
purchase_moved(inventory)
with on_day(DAY2):
before = len(inventory.reserve_bodies)
must(api("POST", f"/sales-orders/{victim['sales_order_id']}/cancel", ADMIN_ROLE, {"reason": "customer changed their mind"})) # the cancel triggers it (right after its response)
order = [sid for sid, _ in inventory.reserve_bodies[before:] if sid != victim["sales_order_id"]]
assert order[:2] == [rush["sales_order_id"], first["sales_order_id"]] # rush first, then first come first served
assert get(rush["sales_order_id"])["schedule_version"] == 2 and get(first["sales_order_id"])["schedule_version"] == 2
sent = len(inventory.reserve_bodies)
assert reschedule()["re_promised"] == [] # the daily run finds the same answer: nothing written
assert len(inventory.reserve_bodies) == sent
def test_SO_36_a_re_promise_racing_a_cancel_of_the_same_order_loses_on_the_order_version_and_writes_nothing(neighbours, inventory, monkeypatch):
from brick_sales_order.application import promising
o = promised_240(inventory, neighbours)
oid = o["sales_order_id"]
purchase_moved(inventory)
real, fired = promising.propose, []
def propose_then_cancel(*a, **kw):
out = real(*a, **kw)
if not fired:
fired.append(1)
monkeypatch.setattr(promising, "propose", real)
must(api("POST", f"/sales-orders/{oid}/cancel", ADMIN_ROLE, {"reason": "gone"})) # another unit of work saves the order first
return out
monkeypatch.setattr(promising, "propose", propose_then_cancel)
with on_day(DAY2):
res = reschedule()
assert res["re_promised"] == [] and res["tasks"] == 0
assert get(oid)["lifecycle"] == "cancelled" and get(oid)["schedule_version"] == 1 and not must(api("GET", "/tasks", REP))["tasks"]
def test_SO_36_a_reservation_inventory_refuses_stays_pending_and_the_next_run_sends_it_again_without_a_new_version(neighbours, inventory):
o = promised_240(inventory, neighbours)
oid = o["sales_order_id"]
purchase_moved(inventory)
inventory.down = True
with on_day(DAY2):
res = reschedule()
assert res["re_promised"] == [oid] and res["reservation_pending"] == 1 and get(oid)["schedule_version"] == 2 # committed first, then asked
inventory.down = False
again = reschedule()
assert again["re_promised"] == [] and again["resent"] == [oid] and again["reservation_pending"] == 0
assert inventory.reserve_bodies[-1][1]["schedule_version"] == 2 and get(oid)["schedule_version"] == 2
def test_SO_36_today_is_the_business_date_not_the_utc_date(neighbours, inventory):
promised_240(inventory, neighbours)
purchase_moved(inventory)
setting("business_timezone", "Pacific/Auckland")
with on_day("2026-11-04T23:30:00+00:00"): # 5 Nov 12:30 in Auckland
assert reschedule()["business_date"] == "2026-11-05"