dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
tests/test_concurrency_bounds.py
"""Anything bounded by a total takes a plain-INSERT claim row (or a version claim) in the same transaction, so two writers on
snapshots that predate each other collide on the key; and every amount, quantity, rate and life is bounded before it can
overflow NUMERIC into a 500."""
from decimal import Decimal as D
import pytest
from conftest import ACCOUNTANT, APPROVER, OWNER, PURCHASING, acquire, api, create_asset, must, on_day, setup_policy
from erp_platform import repository as platform_repository
from sqlalchemy import insert
from sqlalchemy.exc import IntegrityError
from brick_fixed_assets.infrastructure import tables as T
from brick_fixed_assets.infrastructure.tables import Repository
@pytest.fixture
def world(gl):
setup_policy(classes=("VEH",), start="2026-01-01")
return gl
def test_two_partial_retirements_that_together_exceed_100_percent_cannot_both_post(world):
a = create_asset("van")
must(acquire(a["asset_id"], "12000.00", value_date="2026-10-20", ref="pur:V-1:1"))
with on_day("2026-10-31T09:00:00+00:00"):
must(api("POST", "/depreciation-runs", ACCOUNTANT, {"company_code": "UK01", "period_id": "2026-10"}))
run = must(api("GET", "/depreciation-runs", ACCOUNTANT, {"company_code": "UK01"}))["runs"][0]
must(api("POST", f"/depreciation-runs/{run['run_id']}/approve", APPROVER))
with on_day("2026-11-02T09:00:00+00:00"):
first = must(api("POST", f"/assets/{a['asset_id']}/retirements", ACCOUNTANT, {"retirement_date": "2026-11-01", "percent": "50"}))
with on_day("2026-11-02T09:00:01+00:00"):
second = must(api("POST", f"/assets/{a['asset_id']}/retirements", ACCOUNTANT, {"retirement_date": "2026-11-01", "percent": "50"}))
s, b = api("POST", f"/assets/{a['asset_id']}/retirements", ACCOUNTANT, {"retirement_date": "2026-11-01", "percent": "1"})
assert s == 409, b # 50 + 50 are already claimed: nothing is left to propose
must(api("POST", f"/asset-transactions/{first['transaction_id']}/approve", APPROVER))
s, b = api("POST", f"/asset-transactions/{second['transaction_id']}/approve", APPROVER)
assert s == 409 and "asset_unchanged" in str(b), b # the asset's version is the claim: the loser re-reads
v = next(x for x in must(api("GET", f"/assets/{a['asset_id']}", ACCOUNTANT))["values"] if x["area_code"] == "01")
assert D(v["cost"]) == D("6000.00") # never more than 100% retired
def test_two_writers_on_one_asset_version_collide_and_the_second_loses(engine, world):
a = create_asset("van")
with engine.begin() as c:
stale = Repository(c).asset(a["asset_id"]) # what the second writer read
mine = Repository(c).asset(a["asset_id"])
with engine.begin() as c:
Repository(c).save_asset(mine) # the first writer commits and bumps the version
with pytest.raises((platform_repository.Conflict, IntegrityError)):
with engine.begin() as c:
Repository(c).save_asset(stale) # the second still holds the old version: it loses
def test_a_period_is_posted_at_most_once_per_asset_area_and_run(engine):
row = {"asset_id": "A", "area_code": "01", "period_id": "2026-10", "run_id": "UK01:2026-10:1", "amount": D("200.00")}
with engine.begin() as c:
c.execute(insert(T.fa_dep_posting).values(**row))
with pytest.raises(IntegrityError):
with engine.begin() as c: # a second approver whose snapshot predates the first post
c.execute(insert(T.fa_dep_posting).values(**row))
def test_two_runs_for_one_period_collide_on_the_run_key(engine):
row = {"company_code": "UK01", "period_id": "2026-10", "run_no": 1}
with engine.begin() as c:
c.execute(insert(T.fa_dep_run).values(run_id="UK01:2026-10:1", status="simulated", **row))
with pytest.raises(IntegrityError):
with engine.begin() as c:
c.execute(insert(T.fa_dep_run).values(run_id="UK01:2026-10:1b", status="simulated", **row))
def test_two_acquisitions_with_one_reference_collide_on_its_claim(engine):
row = {"acquisition_ref": "pur:B-1:1", "transaction_id": "t1", "amount": D("1.00")}
with engine.begin() as c:
c.execute(insert(T.fa_acquisition_claim).values(**row))
with pytest.raises(IntegrityError):
with engine.begin() as c:
c.execute(insert(T.fa_acquisition_claim).values(**{**row, "transaction_id": "t2"}))
# ── bounds: a value that does not fit is a 422 with the field named, never a 500 ───────────────
HUGE = "1" + "0" * 20 # 21 integer digits: past NUMERIC(18,2)
def bad(status, body, field):
assert status == 422, (status, body)
assert field in str(body), body
def test_amounts_quantities_rates_and_lives_are_bounded_before_they_are_stored(world):
a = create_asset("van")
s, b = acquire(a["asset_id"], HUGE)
bad(s, b, "amount")
s, b = api("POST", "/acquisitions", PURCHASING,
{"acquisition_ref": "pur:B-9:1", "source_kind": "purchase_bill", "company_code": "UK01", "bill_id": "B-9",
"line_no": 1, "value_date": "2026-10-20", "amount": "NaN", "currency": "GBP", "asset_id": a["asset_id"]})
bad(s, b, "amount")
s, b = api("POST", "/acquisitions", PURCHASING,
{"acquisition_ref": "pur:B-9:2", "source_kind": "purchase_bill", "company_code": "UK01", "bill_id": "B-9",
"line_no": 2, "value_date": "2026-10-20", "amount": "1.005", "currency": "GBP", "asset_id": a["asset_id"]})
assert s == 422, (s, b) # more decimals than a minor unit
s, b = api("POST", "/assets", ACCOUNTANT, {"request_id": "q", "company_code": "UK01", "class_code": "VEH", "description": "d",
"quantity": "1" + "0" * 16})
bad(s, b, "quantity")
s, b = api("POST", "/settings/keys", OWNER, {"key_code": "BAD", "method": "declining_balance", "rate_pct": "100.01"})
assert s == 422, (s, b)
s, b = api("POST", "/settings/keys", OWNER, {"key_code": "BAD", "method": "straight_line", "life_periods": 1201})
bad(s, b, "life_periods")
s, b = api("POST", "/settings/keys", OWNER, {"key_code": "BAD", "method": "straight_line", "life_periods": 0})
bad(s, b, "life_periods")
s, b = api("POST", "/settings/ca-rates", OWNER, {"rate_key": "wda_main_pct", "value": "100.5"})
assert s in (409, 422), (s, b)
for field, body in (("proceeds", {"retirement_date": "2026-11-01", "percent": "100", "proceeds": HUGE}),
("percent", {"retirement_date": "2026-11-01", "percent": HUGE}),
("amount", {"retirement_date": "2026-11-01", "amount": HUGE})):
s, b = api("POST", f"/assets/{a['asset_id']}/retirements", ACCOUNTANT, body)
bad(s, b, field)
s, b = api("POST", f"/assets/{a['asset_id']}/revaluations", ACCOUNTANT,
{"area_code": "30", "revalued_amount": HUGE, "valuation_date": "2026-11-01", "evidence_sha256": "a" * 64})
bad(s, b, "revalued_amount")
s, b = api("POST", f"/assets/{a['asset_id']}/impairments", ACCOUNTANT,
{"amount": HUGE, "review_date": "2026-11-01", "evidence_sha256": "a" * 64})
bad(s, b, "amount")
s, b = api("POST", f"/assets/{a['asset_id']}/manual-acquisitions", ACCOUNTANT,
{"request_id": "m", "value_date": "2026-11-01", "amount": HUGE, "offset_account_id": 1})
assert s in (404, 422), (s, b)
def test_two_writers_allocating_the_same_sub_asset_or_run_number_collide_on_the_claim(engine):
"""sub_no and run_no are picked from the rows a snapshot sees; the claim row keyed on the number makes the loser collide."""
from brick_fixed_assets.infrastructure.tables import Repository
for kind, key in (("SUB", "UK01-2026-VEH-0001-2"), ("RUN", "UK01:2026-10:2")):
with engine.begin() as c:
Repository(c).claim_slot(kind, key, "a")
with pytest.raises(IntegrityError):
with engine.begin() as c: # a second snapshot that predates the first claim
Repository(c).claim_slot(kind, key, "b")
assert Repository.first_free([1, 2, 4]) == 5 # never fills the gap below the high-water mark