dodil/erp-brick-inventory

DODIL ERP brick: inventory

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 documents or moves. Counted at the engine, one request at a time (erp_platform.testing.count_statements)."""
import pytest
import world
from conftest import OPERATOR, SYSTEM, api, must
from erp_platform.testing import count_statements
from test_issues import gloves, issue


@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 _posted(n):
    for _ in range(n):
        plan = must(issue(qty="1"))
        must(api("POST", f"/goods-issues/{plan['document_id']}/post", OPERATOR, {}))


def test_posting_an_issue_does_not_grow_with_the_number_of_earlier_issues(counted):
    gloves()
    _posted(2)
    plan = must(issue(qty="1"))
    with counted() as few:
        must(api("POST", f"/goods-issues/{plan['document_id']}/post", OPERATOR, {}))
    _posted(8)
    plan = must(issue(qty="1"))
    with counted() as many:
        must(api("POST", f"/goods-issues/{plan['document_id']}/post", OPERATOR, {}))
    _show("post issue (2 earlier)", few)
    _show("post issue (10 earlier)", many)
    assert many.statements <= few.statements and few.statements <= 29, (few.statements, many.statements)


def test_planning_an_issue_and_reading_stock_do_not_grow_with_the_number_of_moves(counted):
    gloves()
    _posted(2)
    with counted() as few_plan:
        must(issue(qty="1"))
    with counted() as few_stock:
        must(api("GET", "/stock/GLOVES", OPERATOR))
    _posted(8)
    with counted() as many_plan:
        must(issue(qty="1"))
    with counted() as many_stock:
        must(api("GET", "/stock/GLOVES", OPERATOR))
    _show("plan issue (2 moves)", few_plan)
    _show("plan issue (10 moves)", many_plan)
    _show("stock (2 moves)", few_stock)
    _show("stock (10 moves)", many_stock)
    assert many_plan.statements <= few_plan.statements and few_plan.statements <= 7, (few_plan.statements, many_plan.statements)
    assert many_stock.statements <= few_stock.statements and few_stock.statements <= 8, (few_stock.statements, many_stock.statements)


def test_a_scanned_receipt_does_not_grow_with_the_number_of_earlier_receipts(counted):
    world.catalogue()
    world.warehouse()

    def scan(k):
        world.purchase(f"PO-{k}", "COLA-075", "10", "PK24", "21.60")
        return api("POST", "/receipts/scan", OPERATOR, {"purchase_order_id": f"PO-{k}", "delivery_note": f"DN-{k}", "location_id": "WH1/A/01",
                                                         "lines": [{"line_no": 1, "quantity": "8", "unit_code": "PK24"}]})
    for k in range(2):
        must(scan(k))
    with counted() as few:
        must(scan(100))
    for k in range(10, 18):
        must(scan(k))
    with counted() as many:
        must(scan(101))
    _show("scan receipt (2 earlier)", few)
    _show("scan receipt (10 earlier)", many)
    assert many.statements <= few.statements and few.statements <= 38, (few.statements, many.statements)


def _sale(order, n_lines, counted):
    """An order of n lines of COLA-075, held, delivered and validated whole: (reserve, create, validate) statement counts."""
    import fakes
    lines = {k: {"product_id": "COLA-075", "unit_code": "C62", "quantity": "4", "unit_numerator": 3, "unit_denominator": 4}
             for k in range(1, n_lines + 1)}
    fakes.add_order(order, lines)
    with counted() as reserve:
        must(api("POST", f"/reservations/sales-orders/{order}", SYSTEM, {"revision": 1, "warehouse_id": "WH1",
                                                                          "lines": [{"line_no": k, **ln} for k, ln in lines.items()]}))
    with counted() as create:
        doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": order, "warehouse_id": "WH1"}))["document_id"]
    with counted() as validate:
        must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [
            {"line_no": k, "location_id": "WH1/A/01", "quantity": "4", "unit_code": "C62"} for k in lines]}))
    return reserve.statements, create.statements, validate.statements


def test_a_sale_through_the_warehouse_stays_within_budget(counted):
    world.catalogue()
    world.warehouse()
    world.receive("COLA-075", "40", "PK24", price="21.60", location="WH1/A/01")
    _sale("SO-W", 1, counted)                       # the first sale: the customer location now holds stock rows too
    two = _sale("SO-A", 2, counted)
    twenty = _sale("SO-B", 20, counted)
    for k in range(48):
        _sale(f"SO-X{k}", 1, counted)
    later = _sale("SO-C", 2, counted)
    print("BUDGET reserve, create, validate (2 lines):", two, "(20 lines):", twenty, "(2 lines, 50 orders):", later)
    assert two == twenty == later, (two, twenty, later)
    # was (2 lines: 21, 9, 41 / 20 lines: 148, 27, 167): reservation rows, stock and the reserved sum read per line, one INSERT per
    # claim, reservation and valuation row, a DELETE per picked line, a SELECT of the stored keys before each staged upsert
    # 0.1.2: a document or line minted in the request, or read by it, is staged with stored=True/False, so the flush writes it without
    # the key lookup (create 8 -> 6, validate 33 -> 30 measured 2026-10-25; budgets +10%)
    assert two[0] <= 12 and two[1] <= 7 and two[2] <= 33, two