dodil/erp-brick-payments

DODIL ERP brick: payments

tests/test_statement_budget.py

"""Statement budgets: every statement is a round trip (100-300 ms on the live pg wire), so a request's statements must not grow with the number of
unrelated rows. Counted at the engine, one request at a time (erp_platform.testing.count_statements)."""
import pytest
from conftest import CASHIER, allocate, api, bank, must, receipt
from erp_platform.testing import count_statements


@pytest.fixture
def counted(engine):
    return lambda: count_statements(engine)


def _show(label, c):
    print(f"BUDGET {label}: statements={c.statements} round_trips={c.round_trips}")


def test_the_receipt_list_does_not_grow_with_the_number_of_customers(nb, counted):
    for bp in range(1, 3):
        receipt(bp_id=bp, amount="10.00", reference=f"R{bp}")
    with counted() as few:
        must(api("GET", "/receipts", CASHIER))
    for bp in range(3, 11):
        receipt(bp_id=bp, amount="10.00", reference=f"R{bp}")
    with counted() as many:
        must(api("GET", "/receipts", CASHIER))
    _show("receipt list (2 customers)", few)
    _show("receipt list (10 customers)", many)
    assert many.statements <= few.statements and few.statements <= 4, (few.statements, many.statements)       # was 5 and 21


def _cheque_batch(nb, n, first=1):
    bank("BANK1", gl="1200", account_number="12345678", purposes="receive,pay")
    ids = []
    for k in range(first, first + n):
        nb.invoice(f"I{k}", 7311, "100.00", due_date="2026-11-25")
        r = receipt(amount="100.00", received_on="2026-11-18", reference=f"CHQ-{k}", method="cheque", bank_account_id="BANK1")
        must(allocate(r["payment_id"], [(f"I{k}", "100.00")]))
        ids.append(r["payment_id"])
    b = must(api("POST", "/deposit-batches", CASHIER, {"bank_account_id": "BANK1"}))["batch_id"]
    for pid in ids:
        must(api("POST", f"/deposit-batches/{b}/cheques", CASHIER, {"payment_id": pid}))
    return b


def test_closing_a_deposit_batch_does_not_grow_with_the_number_of_cheques(nb, counted):
    b = _cheque_batch(nb, 2)
    with counted() as few:
        must(api("POST", f"/deposit-batches/{b}/close", CASHIER, {"slip_no": "S1"}))
    b = _cheque_batch(nb, 8, first=10)
    with counted() as many:
        must(api("POST", f"/deposit-batches/{b}/close", CASHIER, {"slip_no": "S2"}))
    _show("close batch (2 cheques)", few)
    _show("close batch (8 cheques)", many)
    assert many.statements <= few.statements and few.statements <= 16, (few.statements, many.statements)      # was 17 and 29


def test_allocating_a_receipt_does_not_grow_with_the_customers_history(nb, counted):
    def one(k):
        nb.invoice(f"A{k}", 7311, "100.00", due_date="2026-11-25")
        return receipt(amount="100.00", reference=f"ALLOC-{k}")["payment_id"]
    for k in range(2):
        must(allocate(one(k), [(f"A{k}", "100.00")]))
    pid = one(50)
    with counted() as few:
        must(allocate(pid, [("A50", "100.00")]))
    for k in range(10, 18):
        must(allocate(one(k), [(f"A{k}", "100.00")]))
    pid = one(51)
    with counted() as many:
        must(allocate(pid, [("A51", "100.00")]))
    _show("allocate (2 earlier)", few)
    _show("allocate (10 earlier)", many)
    assert many.statements <= few.statements and few.statements <= 21, (few.statements, many.statements)


def test_a_receipt_and_its_allocation_are_flat_in_lines_and_in_other_receipts(nb, counted):
    """The sale's cash steps: record_receipt, then allocate it over the customer's invoices (2 vs 20 invoices, 2 vs 50 other receipts)."""
    seq = iter(range(10_000))

    def cash(n):
        k = next(seq)
        ids = [f"S{k}-{i}" for i in range(n)]
        for i in ids:
            nb.invoice(i, 7311, "10.00", due_date="2026-11-25")
        with counted() as rec:
            pid = receipt(amount=f"{10 * n}.00", reference=f"CASH-{k}")["payment_id"]
        with counted() as alloc:
            must(allocate(pid, [(i, "10.00") for i in ids]))
        return rec.statements, alloc.statements
    cash(1)                                        # the customer's first receipt
    two, twenty = cash(2), cash(20)
    for _ in range(48):
        cash(1)
    later = cash(2)
    print("BUDGET receipt, allocate: 2 invoices", two, "20 invoices", twenty, "2 invoices after 50 receipts", later)
    assert two == twenty == later, (two, twenty, later)
    # was (16, 24) at 2 invoices and (16, 96) at 20: settings and the zone read three times, three max() reads for 'today', one read of
    # the next seq, one claim INSERT and one allocation INSERT per invoice
    assert two[0] <= 11 and two[1] <= 15, two