dodil/erp-brick-ar

DODIL ERP brick: accounts receivable

tests/test_rules.py

"""The rules, with no database and no HTTP: one test per sentence an accountant would agree to."""
import pathlib
import re
from datetime import date, datetime
from decimal import Decimal as D

import pytest

from brick_ar.domain import numbering
from brick_ar.domain.credit_note import CreditNote
from brick_ar.domain.dates import add_months, check_dated, due_date, end_of_month
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.invoice import Invoice, Line, deposit_deduction_lines
from brick_ar.domain.migration_run import MigrationCheck, MigrationRun
from brick_ar.domain.money import line_net
from brick_ar.domain.readiness import GatesUnmet
from brick_ar.domain.schedule import Schedule
from brick_ar.domain.tax import TaxedLine, TaxQuote, TaxRowIn
from brick_ar.domain.values import Settings, check_setting
from brick_ar.domain.writeoff import Writeoff

NOW = datetime(2026, 11, 20, 9, 0)
TODAY = date(2026, 11, 20)


def quote_for(inv, rates):
    """A tax module's answer for an invoice whose lines are (code by index); rounded per line."""
    lines, rows = [], {}
    for ln in inv.lines:
        code = rates[(ln.line_no - 1) % len(rates)][0]
        pct = rates[(ln.line_no - 1) % len(rates)][1]
        tax = (ln.net_amount * pct / 100).quantize(D("0.01"))
        lines.append(TaxedLine(ln.line_no, code, pct, tax))
        r = rows.setdefault(code, [pct, D(0), D(0)])
        r[1] += ln.net_amount
        r[2] += tax
    return TaxQuote(tuple(lines), tuple(TaxRowIn(c, v[0], v[1], v[2]) for c, v in rows.items()), "per_line")


def draft(lines, **kw):
    return Invoice.raise_draft("man:1", by="b", bp_id=1, currency="GBP", source_kind="manual", source_id="man:1", lines=lines, **kw)


def L(qty="1", price="10.00", disc="0", **kw):
    return Line(line_no=0, description="x", quantity=D(qty), unit_price=D(price), discount_pct=D(disc), **kw)


def issue(inv, quote=None):
    inv.issue(today=TODAY, number="INV-2026-0001", series_id="INV-2026", due_date=date(2026, 12, 20), bill_to_json="{}", terms_id="NET30",
              quote=quote or quote_for(inv, [("S", D(20))]), by="b")


# ── the invariants ────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("mix", [[("S", D(20))], [("S", D(20)), ("Z", D(0))], [("S", D(20)), ("R", D(5)), ("Z", D(0)), ("E", D(0))]])
@pytest.mark.parametrize("prices", [("33.33", "0.01", "999.99"), ("0.005", "1.005", "2.675"), ("100", "100", "100")])
def test_subtotal_plus_tax_equals_total_at_every_rate_mix(mix, prices):
    inv = draft([L("3", p, disc=d) for p, d in zip(prices, ("0", "12.5", "33.3333"))])
    inv.apply_tax(quote_for(inv, mix))
    assert inv.subtotal + inv.tax_total == inv.total
    assert sum(r.tax_amount for r in inv.tax) == inv.tax_total
    assert sum(r.net_amount for r in inv.tax) == inv.subtotal


def test_a_tax_summary_that_does_not_cover_the_lines_is_refused():
    inv = draft([L(), L()])
    q = quote_for(inv, [("S", D(20))])
    bad = TaxQuote(q.lines, (TaxRowIn("S", D(20), D("5.00"), D("1.00")),), "x")
    with pytest.raises(RuleBroken, match="does not cover"):
        inv.apply_tax(bad)


def test_line_net_is_rounded_half_up_once():
    assert line_net(D("3"), D("0.335"), D(0)) == D("1.01")
    assert line_net(None, D("5"), D(0)) == D("0.00")


def test_an_issued_invoice_refuses_every_edit():
    inv = draft([L()])
    issue(inv)
    for call in (lambda: inv.edit(by="b", lines=[L()]), lambda: inv.edit(by="b", customer_po_ref="x"), lambda: inv.cancel(by="b"),
                 lambda: inv.fill_usage(1, D(1), by="b"), lambda: inv.review(by="b", at=NOW), lambda: inv.add_finding("s", "m", "d", "f", "e")):
        with pytest.raises(RuleBroken, match="credit it instead"):
            call()
    assert [type(e).__name__ for e in inv.events] == ["InvoiceRaised", "InvoiceIssued"]


def test_issue_stamps_the_dates_and_freezes_the_tax_point_to_the_source():
    inv = draft([L()], tax_point_date=date(2026, 11, 3))
    issue(inv)
    assert (inv.issue_date, inv.tax_point_date, inv.due_date, inv.lifecycle) == (TODAY, date(2026, 11, 3), date(2026, 12, 20), "issued")
    other = draft([L()])
    issue(other)
    assert other.tax_point_date == TODAY


def test_a_usage_line_with_no_quantity_blocks_issue_and_a_filled_one_does_not():
    inv = draft([Line(line_no=0, description="calls", quantity=None, unit_price=D("2.50"), usage=True)])
    assert [g.gate for g in inv.local_gates() if not g.met] == ["usage_filled"]
    with pytest.raises(GatesUnmet, match="usage_filled"):
        issue(inv)
    inv.fill_usage(1, D(10), by="b")
    assert inv.subtotal == D("25.00") and all(g.met for g in inv.local_gates())
    with pytest.raises(RuleBroken, match="not a usage line"):
        draft([L()]).fill_usage(1, D(1), by="b")


def test_an_invoice_with_no_lines_cannot_issue():
    with pytest.raises(GatesUnmet, match="has_lines"):
        issue(draft([]))


def test_void_needs_an_issued_invoice_and_a_reason_and_keeps_the_number():
    inv = draft([L()])
    with pytest.raises(RuleBroken, match="only an issued"):
        inv.void(today=TODAY, reason="x", by="a")
    issue(inv)
    with pytest.raises(RuleBroken, match="needs a reason"):
        inv.void(today=TODAY, reason=" ", by="a")
    inv.void(today=TODAY, reason="error", by="a")
    assert inv.lifecycle == "void" and inv.invoice_no == "INV-2026-0001"
    with pytest.raises(RuleBroken, match="void"):
        inv.edit(by="a", lines=[L()])


def test_a_deposit_is_deducted_per_tax_code_from_the_final_invoice():
    dep = draft([L("1", "5000.00")], kind="deposit")
    issue(dep)
    lines = deposit_deduction_lines(dep, 2)
    assert [(ln.line_no, ln.tax_code, ln.unit_price, ln.deducts_invoice_id) for ln in lines] == [(2, "S", D("-5000.00"), "man:1")]
    final = draft([L("1", "10000.00")], kind="final")
    final.lines += lines
    final._renumber_and_price()
    final.apply_tax(quote_for(final, [("S", D(20))]))
    assert (final.subtotal, final.tax_total, final.total) == (D("5000.00"), D("1000.00"), D("6000.00"))
    with pytest.raises(RuleBroken, match="issued deposit"):
        deposit_deduction_lines(draft([L()], kind="deposit"), 1)


def test_a_migrated_invoice_is_history_and_is_never_sent():
    inv = draft([L()])
    issue(inv)
    inv.source_kind = "odoo"
    with pytest.raises(RuleBroken, match="history"):
        inv.record_send(at=NOW, address="a@b", channel="email", by="b")


def test_an_export_batch_takes_a_document_once():
    inv = draft([L()])
    inv.mark_exported("2026-10:csv")
    with pytest.raises(RuleBroken, match="already in batch"):
        inv.mark_exported("2026-11:csv")


def test_findings_are_decided_once_by_a_person_with_a_reason_when_dismissed():
    inv = draft([L()])
    f = inv.add_finding("ai_check_draft", "m", "d", "an unusual amount", "2.1x")
    assert f.run_no == 1 and inv.add_finding("ai_check_draft", "m", "d", "again", "e").run_no == 2
    with pytest.raises(RuleBroken, match="needs a reason"):
        inv.decide_finding("ai_check_draft", 1, "dismissed", "", by="b", at=NOW)
    with pytest.raises(RuleBroken, match="accepted or dismissed"):
        inv.decide_finding("ai_check_draft", 1, "open", "", by="b", at=NOW)
    inv.decide_finding("ai_check_draft", 1, "accepted", "", by="b", at=NOW)
    with pytest.raises(RuleBroken, match="already accepted"):
        inv.decide_finding("ai_check_draft", 1, "dismissed", "x", by="c", at=NOW)
    assert inv.lines[0].net_amount == D("10.00")                       # nothing in the draft moved


# ── credit notes and write-offs ──────────────────────────────────────────────────────────────

def test_a_credit_note_needs_lines_stays_within_the_invoice_and_a_second_person():
    cn = CreditNote.draft("cn:1", by="alice", bp_id=1, currency="GBP", reason="goodwill", lines=[L("1", "100.00")], invoice_id="i")
    cn.apply_tax(quote_for(cn, [("S", D(20))]))
    assert cn.total == D("120.00")
    gates = cn.gates(remaining_creditable=D("100.00"), second_approver=True, approver="alice")
    assert {g.gate for g in gates if not g.met} == {"within_invoice", "second_person"}
    with pytest.raises(GatesUnmet):
        cn.approve(today=TODAY, number="CN-1", series_id="CN", by="alice", at=NOW, quote=None, gates=gates)
    ok = cn.gates(remaining_creditable=D("120.00"), second_approver=True, approver="bob")
    cn.approve(today=TODAY, number="CN-2026-0001", series_id="CN-2026", by="bob", at=NOW, quote=None, gates=ok)
    assert (cn.status, cn.credit_date) == ("approved", TODAY)
    with pytest.raises(RuleBroken, match="approved"):
        cn.cancel(by="x")
    with pytest.raises(RuleBroken, match="needs a reason"):
        CreditNote.draft("cn:2", by="a", bp_id=1, currency="GBP", reason=" ", lines=[])
    assert not CreditNote.draft("cn:3", by="a", bp_id=1, currency="GBP", reason="r", lines=[]).gates(
        remaining_creditable=None, second_approver=False, approver="b")[0].met


def test_a_writeoff_gates_outstanding_threshold_and_second_person():
    w = Writeoff.propose("wo:1", by="alice", invoice_id="i", amount=D("30.00"), reason="gone")
    g = w.gates(outstanding=D("20.00"), threshold=25, confirmed=False, second_approver=True, approver="alice")
    assert {x.gate for x in g if not x.met} == {"outstanding", "threshold", "second_person"}
    g = w.gates(outstanding=D("30.00"), threshold=25, confirmed=True, second_approver=True, approver="bob")
    w.approve(today=TODAY, by="bob", at=NOW, gates=g)
    assert w.status == "approved" and w.events[0].amount == D("30.00")
    with pytest.raises(RuleBroken, match="already approved"):
        w.reject(reason="x", by="b", at=NOW)
    below = Writeoff.propose("wo:2", by="a", invoice_id="i", amount=D("12.40"), reason="r")
    assert all(x.met for x in below.gates(outstanding=D("12.40"), threshold=25, confirmed=False, second_approver=False, approver="a"))
    with pytest.raises(RuleBroken):
        Writeoff.propose("wo:3", by="a", invoice_id="i", amount=D("0"), reason="r")


# ── dated rows, terms, numbers ────────────────────────────────────────────────────────────────

def test_a_dated_row_starts_after_today_except_the_first_of_its_key():
    check_dated("x", date(2026, 11, 21), TODAY, first=False)
    check_dated("x", TODAY, TODAY, first=True)
    check_dated("x", date(2026, 12, 1), TODAY, first=True)
    for vf, first in ((TODAY, False), (date(2026, 11, 19), False), (date(2026, 11, 19), True)):
        with pytest.raises(RuleBroken):
            check_dated("x", vf, TODAY, first=first)


def test_due_dates_and_months():
    assert due_date(date(2026, 11, 20), 30, "invoice_date") == date(2026, 12, 20)
    assert due_date(date(2026, 11, 20), 30, "end_of_month") == date(2026, 12, 30)
    assert end_of_month(date(2028, 2, 3)) == date(2028, 2, 29)
    assert add_months(date(2026, 1, 31), 1) == date(2026, 2, 28) and add_months(date(2026, 11, 20), 3) == date(2027, 2, 20)


def test_number_formats_series_and_carried_numbers():
    assert numbering.render("INV-{yyyy}-{seq:4}", 2026, 7) == "INV-2026-0007"
    assert numbering.render("INV/{yyyy}/{seq:5}", 2026, 12) == "INV/2026/00012"
    assert numbering.series_id("INV-{yyyy}-{seq:4}", 2026) == "INV-2026"
    assert numbering.series_id("INV-{yyyy}-{seq:4}", 2026, restarts_yearly=False) == "INV"
    assert numbering.credit_format("INV-{yyyy}-{seq:4}") == "CN-{yyyy}-{seq:4}"
    assert numbering.parse_number("INV/2026/00012") == ("INV/2026", 12)
    assert numbering.parse_number("/") is None
    # Odoo's series continues: a carried number and the format's series are the same series
    assert numbering.series_id("INV/{yyyy}/{seq:5}", 2026) == numbering.parse_number("INV/2026/00012")[0]


def test_settings_check_their_kind_and_choices():
    assert Settings.of().void_policy == "never_sent" and Settings.of().auto_issue == frozenset()
    assert Settings.of({"auto_issue_sources": "schedule, sales_order"}).auto_issue == {"schedule", "sales_order"}
    for key, bad in (("void_policy", "sometimes"), ("business_timezone", "Mars/Base"), ("auto_issue_sources", "carrier pigeon"),
                     ("writeoff_threshold", -1), ("second_approver", "yes"), ("invoice_number_format", "INV"), ("nope", 1)):
        with pytest.raises(ValueError):
            check_setting(key, bad)
    assert check_setting("ledger_mode", None) is None


# ── schedules ──────────────────────────────────────────────────────────────────────────────────

def sched(**kw):
    return Schedule.plan("s|1", by="x", bp_id=1, starts_on=date(2026, 10, 1), periods=12, unit_price=D("120"), currency="GBP", **kw)


def test_periods_are_derived_from_the_start_and_never_counted():
    s = sched()
    assert s.bounds(2) == (date(2026, 11, 1), date(2026, 11, 30)) and s.billing_date(12) == date(2027, 9, 1)
    assert s.due_periods(date(2026, 11, 1), {1}) == [2]
    assert s.due_periods(date(2026, 11, 1), {1, 2}) == []
    assert s.due_periods(date(2026, 11, 20), set()) == [1, 2]                  # a start in the past: the missed periods are due now
    assert s.next_due({1, 2}) == date(2026, 12, 1)
    arrears = sched(in_advance=False)
    assert arrears.billing_date(1) == date(2026, 10, 31) and arrears.due_periods(date(2026, 10, 30), set()) == []
    with pytest.raises(RuleBroken, match="periods 1 to 12"):
        s.bounds(13)


def test_anniversary_clamps_the_day_and_a_quarter_is_three_months():
    s = Schedule.plan("s|2", by="x", bp_id=1, starts_on=date(2026, 1, 31), periods=4, unit_price=D(1), currency="GBP", billing_plan="quarterly")
    assert [s.bounds(k)[0] for k in (1, 2, 3)] == [date(2026, 1, 31), date(2026, 4, 30), date(2026, 7, 31)]
    m = Schedule.plan("s|3", by="x", bp_id=1, starts_on=date(2026, 1, 31), periods=3, unit_price=D(1), currency="GBP")
    assert [m.bounds(k)[0] for k in (1, 2, 3)] == [date(2026, 1, 31), date(2026, 2, 28), date(2026, 3, 31)]


def test_month_start_billing_has_a_stub_first_period_that_may_be_prorated():
    s = sched(billing_day="month_start", prorate_first=True)
    s.starts_on = date(2026, 10, 21)
    assert s.bounds(1) == (date(2026, 10, 21), date(2026, 10, 31)) and s.bounds(2)[0] == date(2026, 11, 1)
    assert s.proration(1) == D(11) / D(31) and s.proration(2) == 1
    assert sched().proration(1) == 1


def test_a_schedule_stops_after_the_period_in_progress_and_finishes_when_all_are_raised():
    s = sched()
    s.stop(today=date(2026, 11, 20), reason="ended", by="x")
    assert s.due_periods(date(2027, 3, 1), {1, 2}) == []                      # period 3 starts after the stop
    arrears = sched(in_advance=False)
    arrears.stop(today=date(2026, 11, 20), reason="ended", by="x")
    assert arrears.due_periods(date(2027, 3, 1), {1}) == [2]
    s2 = sched()
    assert s2.finish(raised=set(range(1, 12)), by="x") is False and s2.finish(raised=set(range(1, 13)), by="x") is True
    assert s2.status == "finished" and s2.due_periods(date(2030, 1, 1), set()) == []
    with pytest.raises(RuleBroken, match="repeats|billed once|recurring"):
        Schedule.plan("s|9", by="x", bp_id=1, starts_on=TODAY, periods=1, unit_price=D(1), currency="GBP", billing_kind="one_off")


# ── the migration run ──────────────────────────────────────────────────────────────────────────

def check(key, equal, must="equal", explanation=""):
    return MigrationCheck(run_id="r", check_key=key, slice="all", odoo_value="1", ours_value="1" if equal else "2", equal=equal,
                          must=must, explanation=explanation)


def test_a_run_is_signed_only_when_every_check_is_equal_or_explained():
    run = MigrationRun(run_id="r", source_db="d", cutover_date=TODAY)
    run.loaded(at=NOW, counts={})
    run.reconciled([check("count", True), check("gaps", False, "explained"), check("totals", False, "equal")])
    assert {g.gate for g in run.gates() if not g.met} == {"must_equal", "zero_unexplained"}
    with pytest.raises(RuleBroken, match="cannot be explained away"):
        run.explain("totals", "all", "because", by="o")
    with pytest.raises(RuleBroken, match="nothing to explain"):
        run.explain("count", "all", "because", by="o")
    run.explain("gaps", "all", "test invoices deleted", by="approver")
    run.reconciled([check("count", True), check("gaps", False, "explained"), check("totals", True)])        # a reload keeps the explanation
    assert run.checks[1].explanation == "test invoices deleted" and all(g.met for g in run.gates())
    run.sign(by="owner", at=NOW)
    assert run.status == "signed" and len(run.checksums_json) == 64
    with pytest.raises(RuleBroken, match="new run"):
        run.loaded(at=NOW, counts={})


def test_an_explanation_does_not_survive_a_check_whose_sides_changed():
    run = MigrationRun(run_id="r")
    run.reconciled([check("gaps", False, "explained")])
    run.explain("gaps", "all", "x", by="o")
    changed = check("gaps", False, "explained")
    changed.ours_value = "3"
    run.reconciled([changed])
    assert run.checks[0].explanation == ""


# ── the domain stays pure ──────────────────────────────────────────────────────────────────────

def test_the_domain_imports_no_framework():
    for path in pathlib.Path(__file__).resolve().parents[1].glob("src/brick_ar/domain/*.py"):
        text = path.read_text()
        assert not re.search(r"^\s*(from|import)\s+(sqlalchemy|fastapi|mcp|erp_platform)", text, re.M), path.name