dodil/erp-brick-gl
DODIL ERP brick: general ledger
tests/test_concurrency.py
"""What must hold when writers overlap. SQLite cannot overlap two transactions, so each test makes ONE of them read what the other
had not yet committed (a stale snapshot) - the situation DataK3's snapshot reads create - and checks the claim that decides it: a plain
INSERT of a key the other holds fails, the loser's whole transaction is rolled back and run again, and sees the winner."""
from datetime import datetime
from decimal import Decimal
from conftest import ACCOUNTANT, AUDITOR, POSTER, api, must, refused
from sqlalchemy import text
from world import books, close_through, invoice, line, post
def test_two_postings_claiming_the_same_entry_number_land_distinct_numbers(monkeypatch):
from brick_gl.infrastructure import repositories
books()
assert invoice("INV-1")[1]["entry_no"] == 1
real = repositories.Numbers.claim
seen = {"stale": True}
def stale_claim(self, company_code, book, fiscal_year, journal_id, at, floor=0):
if seen["stale"]:
seen["stale"] = False # this transaction's snapshot predates INV-1: it also reads "nothing claimed yet"
self.add({"company_code": company_code, "book": book, "fiscal_year": fiscal_year, "entry_no": 1, "journal_id": journal_id, "claimed_at": at})
return 1
return real(self, company_code, book, fiscal_year, journal_id, at, floor)
monkeypatch.setattr(repositories.Numbers, "claim", stale_claim)
s, b = invoice("INV-2")
assert s == 200 and b["entry_no"] == 2
monkeypatch.setattr(repositories.Numbers, "claim", real)
nums = sorted(j["entry_no"] for j in must(api("GET", "/journals", ACCOUNTANT))["journals"])
assert nums == [1, 2]
def test_two_postings_claiming_the_same_chain_link_chain_in_order(monkeypatch):
from brick_gl.infrastructure import repositories
books()
invoice("INV-1")
real = repositories.Chain.claim
seen = {"stale": True}
def stale_claim(self, company_code, journal_id, at):
if seen["stale"]:
seen["stale"] = False
self.add({"company_code": company_code, "chain_seq": 1, "journal_id": journal_id, "claimed_at": at})
return 1, "0" * 64
return real(self, company_code, journal_id, at)
monkeypatch.setattr(repositories.Chain, "claim", stale_claim)
assert invoice("INV-2")[0] == 200
monkeypatch.setattr(repositories.Chain, "claim", real)
assert must(api("GET", "/audit/verify-chain", AUDITOR, {}))["met"] is True and must(api("GET", "/audit/verify-chain", AUDITOR, {}))["journals"] == 2
def test_the_same_source_document_posted_twice_at_once_yields_one_journal_and_a_replay(monkeypatch):
from brick_gl.infrastructure import repositories
books()
assert invoice("INV-1")[1]["replay"] is False
real = repositories.Journals.get
seen = {"stale": True}
def stale_get(self, *key):
if seen["stale"]:
seen["stale"] = False # the second caller's snapshot has not seen the first journal yet
return None
return real(self, *key)
monkeypatch.setattr(repositories.Journals, "get", stale_get)
s, b = invoice("INV-1")
monkeypatch.setattr(repositories.Journals, "get", real)
assert s == 200 and b["replay"] is True
assert len(must(api("GET", "/journals", ACCOUNTANT))["journals"]) == 1
def test_a_close_that_won_the_race_refuses_the_posting_that_read_the_period_open(monkeypatch):
"""The poster read 2026-09 as open; the close committed meanwhile. The poster's own write of the period row finds it closed."""
from brick_gl.infrastructure import repositories
books()
close_through("2026-09")
real = repositories.Periods.for_date
def stale_for_date(self, company_code, on):
row = real(self, company_code, on)
return {**row, "status": "open"} if row else row
monkeypatch.setattr(repositories.Periods, "for_date", stale_for_date)
s, b = invoice("INV-LATE", on="2026-09-30")
monkeypatch.setattr(repositories.Periods, "for_date", real)
refused((s, b), 409, "[period_open]", "2026-09")
assert must(api("GET", "/journals", ACCOUNTANT))["journals"] == []
def test_reversing_twice_at_once_posts_one_reversal(monkeypatch):
from brick_gl.infrastructure import repositories
books()
jid = invoice("INV-1")[1]["journal_id"]
must(api("POST", f"/journals/{jid}/reverse-journal", ACCOUNTANT, {"reason": "first"}))
real = repositories.Journals.get
calls = {"n": 0}
def stale_get(self, *key):
calls["n"] += 1
row = real(self, *key)
if row is not None and calls["n"] <= 2 and row["journal_id"] == int(jid):
return {**row, "status": "posted", "reversal_status": "none"} # the second reverser still sees the original as posted
if calls["n"] <= 3 and row is not None and row["reverses_journal_id"] == int(jid):
return None
return row
monkeypatch.setattr(repositories.Journals, "get", stale_get)
s, b = api("POST", f"/journals/{jid}/reverse-journal", ACCOUNTANT, {"reason": "second"})
monkeypatch.setattr(repositories.Journals, "get", real)
assert s == 409 and "[not_reversed]" in b["detail"]
assert len([j for j in must(api("GET", "/journals", ACCOUNTANT))["journals"] if j["reverses_journal_id"] == jid]) == 1
def test_a_balance_is_the_sum_of_its_lines_after_many_postings_and_the_chain_holds(engine):
books()
for i in range(25):
assert post(f"INV-{i}", "issue", f"2026-10-{(i % 20) + 1:02d}", [line(purpose="receivable", debit=f"{10 + i}.00"), line(purpose="revenue", credit=f"{10 + i}.00")])[0] == 200
with engine.begin() as c:
bal = {r[0]: Decimal(str(r[1])) for r in c.execute(text("SELECT account_id, ROUND(SUM(closing), 2) FROM gl_balance GROUP BY account_id"))}
raw = {r[0]: Decimal(str(r[1])) for r in c.execute(text("SELECT account_id, ROUND(SUM(functional_debit - functional_credit), 2) FROM gl_split_line GROUP BY account_id"))}
assert bal == raw and sum(bal.values()) == 0
chain = must(api("GET", "/audit/verify-chain", AUDITOR, {}))
assert chain["met"] is True and chain["journals"] == 25
assert sorted(j["entry_no"] for j in must(api("GET", "/journals", ACCOUNTANT, {"limit": 100}))["journals"]) == list(range(1, 26))
_ = POSTER
def test_a_recurring_run_twice_at_once_posts_once(monkeypatch):
from world import account_id
from brick_gl.infrastructure import repositories
books()
must(api("POST", "/recurring", ACCOUNTANT, {"recurring_id": "R", "first_period": "2026-01", "lines": [
{"account_id": str(account_id("6100")), "debit": "10.00"}, {"account_id": str(account_id("2100")), "credit": "10.00"}]}))
from conftest import ACCOUNTANT2
must(api("POST", "/recurring/R/approve-recurring", ACCOUNTANT2, {}))
must(api("POST", "/periods/2026-10/run-recurring-entries", ACCOUNTANT, {}))
real = repositories.Journals.get
seen = {"stale": True}
def stale_get(self, *key):
if seen["stale"]:
seen["stale"] = False
return None
return real(self, *key)
out_before = len(must(api("GET", "/journals", ACCOUNTANT))["journals"])
monkeypatch.setattr(repositories.Journals, "get", stale_get)
api("POST", "/periods/2026-10/run-recurring-entries", ACCOUNTANT, {})
monkeypatch.setattr(repositories.Journals, "get", real)
assert len(must(api("GET", "/journals", ACCOUNTANT))["journals"]) == out_before
def test_two_posters_of_one_allocation_run_number_collide_on_the_claim(engine):
"""The run number is claimed by a plain INSERT keyed on it: a second poster whose snapshot predates the first collides."""
import pytest
from sqlalchemy.exc import IntegrityError
from brick_gl.infrastructure.tables import Repository
row = {"company_code": "UK01", "book": "allocation-run:C1:1:0L", "fiscal_year": "2026-10", "entry_no": 1, "journal_id": 0,
"claimed_at": datetime(2026, 10, 22, 9, 0)}
with engine.begin() as c:
Repository(c).numbers.add(row)
with pytest.raises(IntegrityError):
with engine.begin() as c:
Repository(c).numbers.add(row)