dodil/erp-brick-fixed-assets

DODIL ERP brick: fixed assets

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 assets. Counted at the engine, one request at a time (erp_platform.testing.count_statements)."""
import pytest
from conftest import ACCOUNTANT, APPROVER2, acquire, api, create_asset, must, setup_policy
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 _assets(first, n):
    for k in range(first, first + n):
        a = create_asset(request_id=f"req-{k}", description=f"Van {k}")
        must(acquire(a["asset_id"], "12000.00", ref=f"pur:B-{k}:1"))


def test_acquiring_an_asset_does_not_grow_with_the_number_of_other_assets(gl, counted):
    setup_policy(classes=("VEH",))
    _assets(1, 2)
    a = create_asset(request_id="req-a")
    with counted() as few:
        must(acquire(a["asset_id"], "5000.00", ref="pur:BA:1"))
    _assets(3, 8)
    a = create_asset(request_id="req-b")
    with counted() as many:
        must(acquire(a["asset_id"], "5000.00", ref="pur:BB:1"))
    _show("acquire (2 other assets)", few)
    _show("acquire (10 other assets)", many)
    assert many.statements <= few.statements and few.statements <= 74, (few.statements, many.statements)       # was 78


def test_a_depreciation_run_reads_its_assets_in_sets(gl, counted):
    setup_policy(classes=("VEH",))
    _assets(1, 2)
    with counted() as few:
        run = must(api("POST", "/depreciation-runs", ACCOUNTANT, {"company_code": "UK01", "period_id": "2026-10"}))
    with counted() as few_approve:
        must(api("POST", f"/depreciation-runs/{run['run_id']}/approve", APPROVER2))
    _assets(3, 18)
    with counted() as many:
        run = must(api("POST", "/depreciation-runs", ACCOUNTANT, {"company_code": "UK01", "period_id": "2026-11"}))
    with counted() as many_approve:
        must(api("POST", f"/depreciation-runs/{run['run_id']}/approve", APPROVER2))
    _show("simulate run (2 assets)", few)
    _show("simulate run (20 assets)", many)
    _show("approve run (2 assets)", few_approve)
    _show("approve run (20 assets)", many_approve)
    # the assets, the posted rows and usage readings of every asset, the configuration keys and the account map are read in sets / once:
    # nothing grows with the number of assets (was 56 and 126 for 8 assets, 26 and 66 now)
    assert many.statements == few.statements and few.statements <= 30, (few.statements, many.statements)
    assert many_approve.statements == few_approve.statements and few_approve.statements <= 70, (few_approve.statements, many_approve.statements)


def _selects(c):
    """The reads: SELECT statements, less the version check the platform's aggregate save makes per saved root (a write-path cost)."""
    return sum(1 for s in c.sql if s.lstrip().upper().startswith("SELECT") and "SELECT fa_asset.version" not in s) + \
        sum(1 for s in c.sql if s.startswith("-- pipeline"))


def test_a_mass_change_reads_its_assets_in_sets(gl, counted):

    setup_policy(classes=("VEH",))

    def mass(first, n, valid_from):
        _assets(first, n)
        with counted() as c:
            must(api("POST", "/asset-classes/VEH/mass-change", ACCOUNTANT, {"valid_from": valid_from, "cost_centre": f"CC{first}"}))
        return c
    few = mass(1, 2, "2026-11-21")

    many = mass(3, 18, "2026-11-22")
    _show("mass change (2 assets) selects", type("C", (), {"statements": _selects(few), "round_trips": 0}))
    _show("mass change (20 assets) selects", type("C", (), {"statements": _selects(many), "round_trips": 0}))
    assert _selects(many) == _selects(few), (_selects(few), _selects(many))