dodil/erp-brick-gl
DODIL ERP brick: general ledger
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). Also the three answers gl owes its callers
by its own signed spec: GET /companies and GET /charts/{id}/accounts are bare lists, GET /migration-map never truncates."""
import pytest
from conftest import ACCOUNTANT, AUDITOR, OWNER, PAYMENTS, api, must
from erp_platform.testing import count_statements
from sqlalchemy import insert
from world import books, invoice
from brick_gl.infrastructure import tables as t
@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 _mapped(engine, n):
with engine.begin() as c:
c.execute(insert(t.gl_migration_map), [{"odoo_db": "acme", "odoo_model": "account.account", "odoo_id": str(i), "natural_key": str(1000 + i),
"run_id": "r"} for i in range(1, n + 1)])
def test_D1_companies_answers_a_bare_list():
books()
got = must(api("GET", "/companies", AUDITOR))
assert isinstance(got, list) and [c["company_code"] for c in got] == ["UK01"]
assert set(got[0]) >= {"company_code", "name", "country", "functional_currency", "chart_id", "bp_id", "financial_year_end_month", "ledgers"}
def test_D2_a_charts_accounts_answer_a_bare_list():
books()
got = must(api("GET", "/charts/MAIN/accounts", PAYMENTS))
assert isinstance(got, list) and got and set(got[0]) == {"account_id", "code", "name", "type", "subtype"}
assert api("GET", "/charts/NOPE/accounts", PAYMENTS)[0] == 404
def test_D3_the_map_route_never_truncates_and_reads_in_one_query(engine, counted):
_mapped(engine, 60)
with counted() as few:
rows = must(api("GET", "/migration-map/acme/account.account", OWNER, {"ids": "1,2,3"}))
assert [r["odoo_id"] for r in rows] == ["1", "2", "3"] and rows[0]["ours"] == "1001"
ids = ",".join(str(1 + i % 60) for i in range(2500)) # 2500 ids (repeats allowed): the old [:2000] cut the tail
with counted() as many:
rows = must(api("GET", "/migration-map/acme/account.account", OWNER, {"ids": ids}))
assert len(rows) == 2500 and rows[-1]["odoo_id"] == str(1 + 2499 % 60)
assert [r["odoo_id"] for r in must(api("GET", "/migration-map/acme/account.account", OWNER, {"ids": "1,2,99"}))] == ["1", "2"] # the unmapped is omitted
s, body = api("GET", "/migration-map/acme/account.account", OWNER, {"ids": ",".join(str(i) for i in range(100000, 102000))})
assert s == 422, (s, body) # too long to be a list of ids: refused, not cut
_show("migration-map (3 ids)", few)
_show("migration-map (2500 ids)", many)
assert few.statements <= 4 and many.statements <= few.statements + 5, (few.statements, many.statements) # was one query per id
def test_the_company_list_and_a_posting_do_not_grow_with_the_other_rows(counted):
books()
for n in range(2):
must(invoice(f"INV-{n}"))
with counted() as few_post:
must(invoice("INV-A"))
with counted() as few_list:
must(api("GET", "/companies", ACCOUNTANT))
for n in range(10, 18):
must(invoice(f"INV-{n}"))
with counted() as many_post:
must(invoice("INV-B"))
with counted() as many_list:
must(api("GET", "/companies", ACCOUNTANT))
_show("post a journal (2 earlier)", few_post)
_show("post a journal (10 earlier)", many_post)
_show("company list", many_list)
assert many_post.statements <= few_post.statements and few_post.statements <= 26, (few_post.statements, many_post.statements)
assert many_list.statements <= few_list.statements
def test_a_source_posting_is_flat_in_its_lines_and_in_other_journals(counted):
"""What every sale step pays the ledger: one POST /journal-entry/journals (ar's issue, inventory's delivery, payments' receipt)."""
from world import account_id, post
books()
def journal(doc, n):
costs = ("6000", "6100", "6300", "6400") # the lines name different accounts: read in one query, not one each
lines = [{"account_id": str(account_id(costs[k % 4])), "debit": "10.00"} for k in range(n - 1)]
lines.append({"account_id": str(account_id("2100")), "credit": f"{10 * (n - 1)}.00"})
with counted() as c:
must(post(doc, "issue", "2026-10-15", lines))
return c.statements
must(invoice("INV-0"))
two, twenty = journal("J-2", 2), journal("J-20", 20)
for n in range(50):
must(invoice(f"INV-X{n}"))
later = journal("J-2B", 2)
print("BUDGET post a journal: 2 lines", two, "20 lines", twenty, "2 lines after 50 journals", later)
assert two == twenty == later, (two, twenty, later)
assert two <= 20, two # was 23 at 2 lines, 26 at 20 (one account read per account): settings read per key, two balance sums