dodil/erp-brick-sales-order

DODIL ERP brick: sales order

tests/test_rules.py

"""The domain rules without a database: one test per sentence a business person would agree to."""
from datetime import date, datetime, timezone
from decimal import Decimal

import pytest

from brick_sales_order.domain import gates as g
from brick_sales_order.domain.errors import GateUnmet, RuleBroken
from brick_sales_order.domain.number_series import NumberSeries
from brick_sales_order.domain.sales_order import LineSpec, SalesOrder
from brick_sales_order.domain.values import money, num, render_number

D = Decimal
TODAY = date(2026, 11, 20)
NOW = datetime(2026, 11, 20, 9, 0, tzinfo=timezone.utc)


def spec(product="P", qty="12", price="10.00", policy="delivered", kind="", **kw):
    return LineSpec(product, product, "C62", D(qty), 1, 1, D(price), D(price), D(0), TODAY, "LIST-GBP row", "standard", policy, kind, "",
                    tax_code="S", **kw)


def order(*specs, confirmed=True, sid="O1"):
    o = SalesOrder.draft(sid, owner="rep@x", bp_id=1, currency="GBP", ordered_on=None, today=TODAY)
    for s in specs or (spec(),):
        o.add_line(s)
    if confirmed:
        o.confirm(gates=g.confirm_gates(o, g.Facts()), order_no="SO-2026-0001", now=NOW, today=TODAY)
    return o


def deliver(o, line, qty, ref, on=TODAY, issued=None, **kw):
    return o.record_delivery(line, D(qty), ref, on, today=TODAY, by="f@x", issued=issued or {}, **kw)


def test_an_order_is_dated_today_never_earlier():
    with pytest.raises(RuleBroken, match="dated today"):
        SalesOrder.draft("O", owner="r", bp_id=1, currency="GBP", ordered_on=date(2026, 11, 19), today=TODAY)
    assert SalesOrder.draft("O", owner="r", bp_id=1, currency="GBP", ordered_on=None, today=TODAY, history=False).ordered_on == TODAY
    assert SalesOrder.draft("O", owner="r", bp_id=1, currency="GBP", ordered_on=date(2025, 1, 1), today=TODAY, history=True).ordered_on == date(2025, 1, 1)


def test_totals_are_the_sum_of_the_lines_and_a_draft_edits_a_confirmed_order_does_not():
    o = order(spec(qty="3", price="1.005"), confirmed=False)
    o.set_tax(1, "S", D("0.60"))
    assert (o.subtotal, o.tax_total, o.total) == (D("3.02"), D("0.60"), D("3.62"))
    o.confirm(gates=g.confirm_gates(o, g.Facts()), order_no="N", now=NOW, today=TODAY)
    for edit in (lambda: o.add_line(spec()), lambda: o.remove_line(1), lambda: o.set_tax(1, "S", D(1)), lambda: o.record_customer_po("PO")):
        with pytest.raises(RuleBroken, match="confirmed: its lines and prices are frozen"):
            edit()


def test_confirming_needs_every_gate_and_names_each_unmet_one():
    o = order(confirmed=False)
    o.lines[0].priced_from, o.lines[0].tax_code = "", ""
    f = g.Facts(customer_active=False, credit={"hold": True, "reason": "75 days", "mode": "block"}, po_required=True)
    with pytest.raises(GateUnmet) as e:
        o.confirm(gates=g.confirm_gates(o, f), order_no="N", now=NOW, today=TODAY)
    assert {x.gate for x in e.value.gates} == {"customer_active", "lines_priced", "lines_taxed", "credit", "po_ref"}
    assert "credit: on hold: 75 days (fix: ask an order admin to override)" in str(e.value)
    assert o.lifecycle == "draft"


def test_an_empty_order_cannot_be_confirmed():
    o = order(confirmed=False)
    o.remove_line(1)
    assert [x.gate for x in g.confirm_gates(o, g.Facts()) if not x.met] == ["lines_priced"]


def test_delivered_never_exceeds_ordered_and_a_reference_counts_once():
    o = order()
    assert deliver(o, 1, 5, "DN-1") is not None and deliver(o, 1, 5, "DN-1") is None
    assert o.delivered(1) == 5 and o.fulfilment_status() == "partly delivered"
    with pytest.raises(RuleBroken, match="would make 13 of 12"):
        deliver(o, 1, 8, "DN-2")
    deliver(o, 1, 7, "DN-3")
    assert o.fulfilment_status() == "delivered"


def test_a_delivery_is_never_future_or_before_the_confirmation_but_an_opening_one_may_be():
    o = order()
    with pytest.raises(RuleBroken, match="future"):
        deliver(o, 1, 1, "A", on=date(2026, 11, 21))
    with pytest.raises(RuleBroken, match="before the confirmation"):
        deliver(o, 1, 1, "B", on=date(2026, 11, 19))
    assert deliver(o, 1, 1, "odoo:9", on=date(2026, 6, 1), kind="opening", source="migration").delivered_on == date(2026, 6, 1)


def test_a_correction_is_a_new_negative_row_against_an_original_and_never_below_invoiced():
    o = order()
    deliver(o, 1, 5, "DN-1")
    c = deliver(o, 1, -2, "DN-1-C", corrects_ref="DN-1", kind="correction")
    assert c.quantity == -2 and o.delivered(1) == 3 and len(o.deliveries) == 2
    with pytest.raises(RuleBroken, match="negative"):
        deliver(o, 1, 1, "X", corrects_ref="DN-1", kind="correction")
    with pytest.raises(RuleBroken, match="below what is already invoiced"):
        deliver(o, 1, -1, "DN-1-D", corrects_ref="DN-1", kind="correction", issued={1: D(3)})


def test_invoiceable_quantity_follows_the_policy_and_never_exceeds_it():
    o = order(spec("A", qty="12", policy="ordered"), spec("B", qty="12", policy="delivered"))
    deliver(o, 2, 5, "DN-1")
    assert (o.due_qty(1, {}), o.due_qty(2, {})) == (12, 5)
    assert (o.due_qty(1, {1: D(5)}), o.due_qty(2, {2: D(5)})) == (7, 0)                    # SO-12: ordered 7 to invoice, delivered 0
    assert o.invoice_status({1: D(5), 2: D(5)}) == "partly invoiced" and o.invoice_status({}) == "to invoice"
    assert o.invoice_status({1: D(12), 2: D(5)}) == "partly invoiced"
    deliver(o, 2, 7, "DN-2")
    assert o.invoice_status({1: D(12), 2: D(12)}) == "invoiced"


def test_asking_invoicing_for_the_same_thing_twice_is_never_possible():
    o = order(spec(qty="4", policy="ordered"))
    plan = o.plan_invoice({})
    o.add_invoice_request(plan["kind"], plan["lines"], plan["net"], D(0), by="b", now=NOW)
    assert o.due_qty(1, {}) == 0                                                              # our own request log counts as invoiced
    with pytest.raises(GateUnmet, match="something_due"):
        o.plan_invoice({})


def test_status_is_derived_and_nothing_stores_it():
    from brick_sales_order.infrastructure.tables import md
    cols = {c.name for n in ("so_order", "so_line", "so_delivery") for c in md.tables[n].columns}
    assert not {"status", "fulfilment_status", "invoice_status", "delivered", "invoiced", "to_invoice", "qty_delivered", "qty_invoiced"} & cols
    assert order(confirmed=False).fulfilment_status() == "not started"


def test_a_deposit_is_deducted_until_used_and_never_exceeds_the_order_total():
    o = order(spec(qty="10", price="1000.00", policy="ordered"))
    o.deposit_pct = D(50)
    amount = o.plan_deposit()
    assert amount == D("5000.00")
    o.add_invoice_request("deposit", [], amount, D(0), by="b", now=NOW)
    with pytest.raises(GateUnmet, match="within_total"):
        o.plan_deposit(D("5000.01"))
    plan = o.plan_invoice({}, deposits=[{"invoice_id": "INV-D1", "request_no": 1, "tax_category": "standard", "amount": amount, "remaining": amount}])
    assert (plan["net"], plan["deduct"], plan["kind"]) == (D("10000.00"), D("5000.00"), "final")
    assert plan["deductions"][0]["deposit_invoice_id"] == "INV-D1" and plan["deductions"][0]["tax_category"] == "standard"
    o.add_invoice_request("final", plan["lines"], plan["net"] - plan["deduct"], plan["deduct"], by="b", now=NOW)
    assert o.unused_deposit() == 0


def test_a_deposit_needs_a_percentage():
    o = order(spec(qty="1", price="100.00", policy="ordered"))
    with pytest.raises(GateUnmet, match="deposit_set"):
        o.plan_deposit()


def test_amendments_never_go_below_done_and_are_never_approved_by_the_amender():
    o = order()
    deliver(o, 1, 4, "DN-1")
    with pytest.raises(GateUnmet, match="not_below_done"):
        o.propose_amendment([{"line_no": 1, "quantity": D(3)}], "less", by="rep", now=NOW, issued={})
    rev = o.propose_amendment([{"line_no": 1, "quantity": D(14)}], "more", by="rep", now=NOW, issued={})
    with pytest.raises(GateUnmet, match="no_pending_amendment"):
        o.propose_amendment([{"line_no": 1, "quantity": D(15)}], "again", by="rep", now=NOW, issued={})
    with pytest.raises(GateUnmet, match="approver_not_amender"):
        o.apply_revision(rev.revision, {1: spec(qty="14")}, by="rep", now=NOW, issued={}, needs_approval=True)
    o.apply_revision(rev.revision, {1: spec(qty="14", price="11.00")}, by="boss", now=NOW, issued={}, needs_approval=True)
    assert (o.revision, o.lines[0].quantity, o.subtotal) == (2, 14, D("154.00"))


def test_an_amendment_needs_a_reason_and_a_change():
    o = order()
    with pytest.raises(RuleBroken, match="reason"):
        o.propose_amendment([{"line_no": 1, "quantity": D(14)}], " ", by="r", now=NOW, issued={})
    with pytest.raises(RuleBroken, match="at least one line"):
        o.propose_amendment([], "x", by="r", now=NOW, issued={})
    with pytest.raises(RuleBroken, match="no line 9"):
        o.propose_amendment([{"line_no": 9, "quantity": D(1)}], "x", by="r", now=NOW, issued={})


def test_cancelling_a_confirmed_order_needs_a_reason_and_nothing_done():
    o = order(spec(policy="ordered"))
    with pytest.raises(RuleBroken, match="reason"):
        o.cancel_order(" ", by="a", issued={})
    plan = o.plan_invoice({})
    o.add_invoice_request("final", plan["lines"], plan["net"], D(0), by="b", now=NOW)
    assert o.cancel_order("withdrew", by="a", issued={}) == ["so:O1:1"] and o.lifecycle == "cancelled"
    p = order(spec(policy="ordered"), sid="O2")
    with pytest.raises(GateUnmet, match="nothing_invoiced"):
        p.cancel_order("x", by="a", issued={1: D(1)})


def test_finishing_locks_once_and_only_when_delivered_and_invoiced_in_full():
    o = order(spec(qty="4"))
    deliver(o, 1, 4, "DN-1")
    assert not o.can_finish({1: D(3)})
    with pytest.raises(RuleBroken, match="not every line"):
        o.finish({1: D(3)})
    assert o.finish({1: D(4)}) is True and o.finish({1: D(4)}) is False and o.lifecycle == "done"
    with pytest.raises(RuleBroken, match="done: raise a new order"):
        o.add_line(spec())


def test_a_repeating_line_is_billed_by_its_schedule_not_by_a_request():
    o = order(spec("HOST", qty="12", kind="recurring", policy="ordered"))
    assert o.due_qty(1, {}) == 0 and not o.can_finish({1: D(12)})
    with pytest.raises(GateUnmet, match="something_due"):
        o.plan_invoice({})


def test_a_suggestion_is_replaced_while_open_and_kept_once_decided():
    o = order(confirmed=False)
    a = o.open_suggestion("ai_order_from_po", model="m", inputs_ref="x", output={"lines": [{"product_id": "P", "quantity": "1"}]})
    b = o.open_suggestion("ai_order_from_po", model="m2", inputs_ref="x", output={"lines": [{"product_id": "P", "quantity": "2"}]})
    assert a is b and len(o.suggestions) == 1 and b.model == "m2" and o.has_open_suggestion()
    res = o.resolve_suggestion("ai_order_from_po", 1, [{"item": "line:0", "action": "reject"}], by="rep", now=NOW)
    o.apply_suggestion("ai_order_from_po", 1, res, [], by="rep", now=NOW)
    assert o.suggestions[0].decision == "rejected" and not o.has_open_suggestion() and len(o.lines) == 1
    c = o.open_suggestion("ai_order_from_po", model="m3", inputs_ref="y", output={"lines": []})
    assert c.suggestion_no == 2                                                              # never over an accepted one
    with pytest.raises(RuleBroken, match="no item line:3"):
        o.resolve_suggestion("ai_order_from_po", 2, [{"item": "line:3", "action": "accept"}], by="r", now=NOW)


def test_the_number_series_never_reuses_or_goes_down():
    s = NumberSeries("SO-2026")
    assert [s.claim(), s.claim()] == [1, 2]
    s.raise_to(40)
    assert s.claim() == 41
    with pytest.raises(RuleBroken, match="never reused"):
        s.raise_to(10)


def test_numbers_and_money_read_as_a_person_reads_them():
    assert render_number("SO-{yyyy}-{seq:4}", 2026, 7) == "SO-2026-0007" and render_number("S{seq:5}", 2026, 42) == "S00042"
    assert (num(D("12.0000")), num(D("24"), 2), num(D("1.2500")), num(D("100"))) == ("12", "24.00", "1.25", "100")
    assert money(D("2.675")) == D("2.68")