dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
tests/test_acquisitions.py
"""FA-06 (purchasing hands a bill line over), FA-07 (manual acquisition, grant), FA-08 (threshold, low-value assets) and
FA-09 (assets under construction settled into the finished assets) — Given/When/Then from the signed spec."""
from decimal import Decimal as D
import pytest
from conftest import ACC, ACCOUNTANT, OWNER, PURCHASING, acquire, api, create_asset, must, on_day, setup_policy
from erp_platform import seam
CLEARING, CARD = 1400, 2210
@pytest.fixture
def world(gl):
setup_policy(classes=("VEH", "IT", "BUILD", "PLANT"))
with on_day("2026-01-01T09:00:00+00:00"):
must(api("POST", "/settings/keys", OWNER, {"key_code": "IMM", "method": "immediate"}))
must(api("POST", "/settings/classes", OWNER, {"class_code": "LVA", "name": "Low value", "is_low_value": True,
"areas": [{"area_code": "01", "key_code": "IMM"}]}))
must(api("POST", "/settings/account-determination", OWNER, {"class_code": "LVA", "area_code": "01", **ACC}))
must(api("POST", "/settings/controls", OWNER, {"setting_key": "asset_clearing_account_id", "value": str(CLEARING)}))
return gl
def cost(asset_id, area):
got = must(api("GET", f"/assets/{asset_id}", ACCOUNTANT))
return next(D(v["cost"]) for v in got["values"] if v["area_code"] == area)
def test_a_bill_line_is_capitalised_once_in_every_area_and_clearing_nets_to_zero(world):
a = create_asset()
s, got = acquire(a["asset_id"], "32000.00", ref="pur:BILL-2026-00031:1")
assert s == 200 and got["replay"] is False, got
assert got["asset_id"] == a["asset_id"] and {r["area_code"] for r in got["areas"]} == {"01", "30", "CA"}
assert all(D(r["cost"]) == D("32000.00") for r in got["areas"])
posted = {r["area_code"]: r for r in got["areas"] if r["journal_id"]}
assert set(posted) == {"01", "30"} # CA is statistical: no journal
assert sorted(j["ledger_id"] for j in world.journals.values()) == ["10", "20"]
for j in world.journals.values():
assert j["journal_date"] == "2026-10-20" # the bill's date, not today's
assert j["journal_id"] == str(seam.derived_id(j["source_document"], j["event"], j["ledger_id"]))
debit = next(ln for ln in j["lines"] if "debit" in ln)
credit = next(ln for ln in j["lines"] if "credit" in ln)
# the class cost account 150 is debited; asset clearing is a PURPOSE (purchasing's account, resolved by gl),
# credited in each ledger to meet purchasing's own bill journal debit there, so clearing nets to zero
assert debit["account_id"] == str(ACC_COST) and D(debit["debit"]) == D("32000.00")
assert credit["purpose"] == "asset_clearing" and D(credit["credit"]) == D("32000.00")
ACC_COST = 150
def test_the_same_acquisition_ref_posts_nothing_new(world):
a = create_asset()
first = must(acquire(a["asset_id"], ref="pur:B-9:1"))
n = len(world.journals)
again = must(acquire(a["asset_id"], ref="pur:B-9:1"))
assert again["replay"] is True and again["transaction_id"] == first["transaction_id"] and len(world.journals) == n
assert cost(a["asset_id"], "01") == D("12000.00")
def test_a_credit_from_the_supplier_reduces_cost_in_every_area(world):
a = create_asset()
must(acquire(a["asset_id"], "12000.00", ref="pur:B-1:1"))
got = must(acquire(a["asset_id"], "-2000.00", ref="pur:B-2:1", source_kind="purchase_credit"))
assert {r["area_code"]: D(r["cost"]) for r in got["areas"]} == {c: D("-2000.00") for c in ("01", "30", "CA")}
assert cost(a["asset_id"], "01") == D("10000.00") and cost(a["asset_id"], "CA") == D("10000.00")
credit = [j for j in world.journals.values() if j["source_document"].endswith("B-2:1")]
assert len(credit) == 2 and all(next(ln for ln in j["lines"] if ln.get("purpose") == "asset_clearing")
["debit"] == "2000.00" for j in credit) # the mirror: clearing debited, cost credited
def test_a_credit_cannot_take_cost_below_zero_and_a_positive_credit_is_refused(world):
a = create_asset()
must(acquire(a["asset_id"], "100.00", ref="pur:B-1:1"))
s, b = acquire(a["asset_id"], "-500.00", ref="pur:B-2:1", source_kind="purchase_credit")
assert s == 422 and b["detail"]["gate"] == "bounds"
s, b = acquire(a["asset_id"], "5.00", ref="pur:B-3:1", source_kind="purchase_credit")
assert s == 422
def test_an_asset_of_another_company_or_a_retired_one_is_refused_with_its_reason(world):
a = create_asset()
s, b = acquire(a["asset_id"], company_code="UK02", ref="pur:B-5:1")
assert s == 409 and b["detail"]["gate"] == "same_company" and b["detail"]["met"] is False
assert not world.journals
def test_a_closed_ledger_period_refuses_the_hand_over_and_leaves_nothing_behind(world):
a = create_asset()
world.closed.add(("20", "2026-10"))
s, b = acquire(a["asset_id"], ref="pur:B-6:1")
assert s == 409 and b["detail"]["gate"] == "period_open" and "30" in b["detail"]["reason"]
assert not world.journals
world.closed.clear()
assert must(acquire(a["asset_id"], ref="pur:B-6:1"))["replay"] is False # the bill's hand-over is retried later
def test_the_depreciation_start_follows_the_keys_convention(world):
a = create_asset()
must(acquire(a["asset_id"], "12000.00", value_date="2026-10-20", ref="pur:B-7:1"))
got = must(api("GET", f"/assets/{a['asset_id']}", ACCOUNTANT))
assert got["capitalised_on"] == "2026-10-20" or got["capitalised_on"].startswith("2026-10")
def test_a_new_asset_can_be_created_from_the_class_and_description(world):
s, got = acquire(None, ref="pur:B-8:1", class_code="VEH", description="Forklift")
assert s == 200 and got["asset_id"].startswith("UK01-")
def test_a_deactivated_asset_is_refused_for_a_bill(world):
a = create_asset()
must(api("POST", f"/assets/{a['asset_id']}/deactivate", ACCOUNTANT, {"reason": "created by mistake"}))
s, b = acquire(a["asset_id"], ref="pur:B-10:1")
assert s == 409 and b["detail"]["gate"] == "not_retired" and not world.journals
# ── FA-07: a manual acquisition and a grant ─────────────────────────────────────────────────
def manual(asset_id, amount, offset, request_id, who=ACCOUNTANT, **kw):
return api("POST", f"/assets/{asset_id}/acquisitions", who,
{"request_id": request_id, "value_date": "2026-11-18", "amount": amount, "offset_account_id": offset, **kw})
def test_a_card_purchase_offsets_the_named_account_never_asset_clearing(world):
a = create_asset(description="Laptop")
got = must(manual(a["asset_id"], "1450.00", CARD, "m1"))
assert got["status"] == "posted"
assert sorted(j["ledger_id"] for j in world.journals.values()) == ["10", "20"]
for j in world.journals.values():
credit = next(ln for ln in j["lines"] if "credit" in ln)
assert credit["account_id"] == str(CARD) and credit["credit"] == "1450.00" and j["journal_date"] == "2026-11-18"
assert cost(a["asset_id"], "CA") == D("1450.00")
s, b = manual(a["asset_id"], "100.00", CLEARING, "m2")
assert s == 409 and "offset_not_clearing" in str(b)
def test_a_grant_reduces_cost_in_book_and_ifrs_but_not_the_tax_area_unless_told(world):
press = create_asset(description="Press")
must(acquire(press["asset_id"], "40000.00", ref="pur:B-20:1"))
must(manual(press["asset_id"], "-10000.00", 1300, "g1"))
assert [cost(press["asset_id"], c) for c in ("01", "30", "CA")] == [D("30000.00"), D("30000.00"), D("40000.00")]
must(manual(press["asset_id"], "-5000.00", 1300, "g2", tax_area_applies=True))
assert cost(press["asset_id"], "CA") == D("35000.00")
def test_a_manual_acquisition_is_idempotent_dated_a_fact_and_bounded(world):
a = create_asset()
first = must(manual(a["asset_id"], "1450.00", CARD, "m1"))
again = must(manual(a["asset_id"], "1450.00", CARD, "m1"))
assert again["replay"] is True and again["transaction_id"] == first["transaction_id"]
assert cost(a["asset_id"], "01") == D("1450.00")
s, b = manual(a["asset_id"], "100.00", CARD, "m3", value_date="2026-11-21") # tomorrow
assert s == 409 and "not_future" in str(b)
s, b = manual(a["asset_id"], "1" + "0" * 17 + ".00", CARD, "m4") # does not fit NUMERIC(18,2)
assert s == 422 and "amount" in str(b)
s, b = manual(a["asset_id"], "-99999.00", CARD, "m5") # a grant above the cost
assert s == 422
def test_only_the_asset_accountant_posts_a_manual_acquisition(world):
a = create_asset()
s, _ = manual(a["asset_id"], "100.00", CARD, "m1", who=PURCHASING)
assert s == 403
# ── FA-08: the capitalisation threshold and low-value assets ────────────────────────────────
def test_below_the_threshold_the_class_list_tells_purchasing_and_a_manual_acquisition_is_refused(world):
s, classes = api("GET", "/asset-classes", PURCHASING)
assert s == 200
by = {c["class_code"]: c for c in classes["classes"]}
assert by["IT"]["capitalisation_threshold"] == "500.00" and by["IT"]["is_low_value"] is False
assert by["LVA"]["is_low_value"] is True
it = create_asset(cls="IT", description="Monitor")
s, b = manual(it["asset_id"], "300.00", CARD, "m1")
assert s == 409 and "above_threshold" in str(b) and not world.journals
assert must(api("GET", f"/assets/{it['asset_id']}", PURCHASING))["capitalisation_threshold"] == "500.00"
def test_a_low_value_asset_is_depreciated_in_full_in_its_acquisition_period(world):
lva = create_asset(cls="LVA", description="Monitor")
must(manual(lva["asset_id"], "450.00", CARD, "m1")) # 450.00 is under the 500.00 threshold, but the class is LVA
v = next(x for x in must(api("GET", f"/assets/{lva['asset_id']}", ACCOUNTANT))["values"] if x["area_code"] == "01")
assert v["cost"] == "450.00" and v["planned_to_date"] == "450.00" # all of it planned in the period of acquisition
# ── FA-09: assets under construction ────────────────────────────────────────────────────────
@pytest.fixture
def auc(world):
with on_day("2026-01-01T09:00:00+00:00"):
must(api("POST", "/settings/keys", OWNER, {"key_code": "NONE", "method": "none"}))
must(api("POST", "/settings/classes", OWNER, {"class_code": "AUC", "name": "Under construction", "is_auc": True,
"areas": [{"area_code": "01", "key_code": "NONE"},
{"area_code": "30", "key_code": "NONE"}]}))
for code in ("01", "30"):
must(api("POST", "/settings/account-determination", OWNER, {"class_code": "AUC", "area_code": code, **ACC}))
site = create_asset(cls="AUC", description="New warehouse")
for n, (what, amount) in enumerate((("steel", "120000.00"), ("roofing", "60000.00"), ("electrics", "30000.00")), start=1):
must(acquire(site["asset_id"], amount, ref=f"pur:{what}:1"))
must(manual(site["asset_id"], "15000.00", 2500, "labour"))
build, plant = create_asset("r-b", cls="BUILD", description="Warehouse"), create_asset("r-p", cls="PLANT", description="Lifts")
return site["asset_id"], build["asset_id"], plant["asset_id"]
def rule(line, *receivers):
return {"line_item_transaction_id": line, "receivers": [{"asset_id": a, **share} for a, share in receivers]}
def test_the_aucs_line_items_are_settled_into_the_finished_assets_per_the_rule(auc, world):
site, build, plant = auc
lines = must(api("GET", f"/assets/{site}/line-items", ACCOUNTANT))["line_items"]
assert sorted(x["amount"] for x in lines) == ["120000.00", "15000.00", "30000.00", "60000.00"]
assert all(x["unsettled"] == x["amount"] for x in lines)
pct = lambda p: {"percent": p} # noqa: E731
d = must(api("POST", f"/assets/{site}/settlements", ACCOUNTANT, {"rules": [
rule("acq:pur:steel:1", (build, pct("100"))), rule("acq:pur:roofing:1", (build, pct("100"))),
rule("acq:pur:electrics:1", (plant, pct("100"))), rule("man:" + site + ":labour", (build, pct("80")), (plant, pct("20")))]}))
assert d["status"] == "draft"
s = must(api("POST", f"/assets/{site}/settlements/{d['settlement_date']}/{d['seq']}/settle", ACCOUNTANT))
assert s["status"] == "settled" and s["replay"] is False
assert cost(build, "01") == D("192000.00") and cost(build, "30") == D("192000.00")
assert cost(plant, "01") == D("33000.00")
assert cost(site, "01") == D("0.00") # the AuC is emptied
got = must(api("GET", f"/assets/{build}", ACCOUNTANT))
assert got["capitalised_on"] == "2026-11-20" # the settlement date starts it
assert all(x["unsettled"] == "0.00" for x in must(api("GET", f"/assets/{site}/line-items", ACCOUNTANT))["line_items"])
settle = [j for j in world.journals.values() if j["event"] == "settle"]
assert len(settle) == 4 and {j["ledger_id"] for j in settle} == {"10", "20"} # BUILD and PLANT, per posting ledger
for ledger in ("10", "20"):
debits = {ln["asset_id"]: D(ln["debit"]) for j in settle if j["ledger_id"] == ledger
for ln in j["lines"] if "debit" in ln}
credits = sum(D(ln["credit"]) for j in settle if j["ledger_id"] == ledger for ln in j["lines"] if "credit" in ln)
assert debits == {build: D("192000.00"), plant: D("33000.00")} and credits == D("225000.00")
assert all(ln["asset_id"] == site for j in settle if j["ledger_id"] == ledger for ln in j["lines"] if "credit" in ln)
again = must(api("POST", f"/assets/{site}/settlements/{d['settlement_date']}/{d['seq']}/settle", ACCOUNTANT))
assert again["replay"] is True and cost(build, "01") == D("192000.00")
def test_a_line_item_is_settled_once_and_never_beyond_what_is_left(auc):
site, build, plant = auc
one = {"percent": "100"}
must(api("POST", f"/assets/{site}/settlements", ACCOUNTANT, {"rules": [rule("acq:pur:steel:1", (build, one))]}))
second = must(api("POST", f"/assets/{site}/settlements", ACCOUNTANT, {"rules": [rule("acq:pur:steel:1", (plant, one))]}))
# both drafted while the line was open; the first settles, the second must then be refused
s, _ = api("POST", f"/assets/{site}/settlements/{second['settlement_date']}/1/settle", ACCOUNTANT)
assert s == 200
s, b = api("POST", f"/assets/{site}/settlements/{second['settlement_date']}/{second['seq']}/settle", ACCOUNTANT)
assert s == 409 and cost(build, "01") == D("120000.00") and cost(plant, "01") == D("0.00"), (s, b)
s, b = api("POST", f"/assets/{site}/settlements", ACCOUNTANT, {"rules": [rule("acq:pur:steel:1", (plant, one))]})
assert s == 409 # already settled: refused at the draft too
def test_a_partial_settlement_leaves_the_rest_on_the_auc_and_the_shares_must_add_up(auc):
site, build, plant = auc
s, b = api("POST", f"/assets/{site}/settlements", ACCOUNTANT,
{"rules": [rule("acq:pur:roofing:1", (build, {"percent": "60"}), (plant, {"percent": "30"}))]})
assert s == 409 and "100%" in str(b) # fully_distributed
d = must(api("POST", f"/assets/{site}/settlements", ACCOUNTANT,
{"rules": [rule("acq:pur:roofing:1", (build, {"amount": "25000.00"}))]}))
must(api("POST", f"/assets/{site}/settlements/{d['settlement_date']}/{d['seq']}/settle", ACCOUNTANT))
assert cost(build, "01") == D("25000.00") and cost(site, "01") == D("200000.00") # 225,000 collected, 25,000 moved
s, _ = api("POST", f"/assets/{site}/settlements", ACCOUNTANT,
{"rules": [rule("acq:pur:roofing:1", (plant, {"amount": "35000.01"}))]})
assert s == 409 # more than the 35,000.00 left
rest = must(api("POST", f"/assets/{site}/settlements", ACCOUNTANT,
{"rules": [rule("acq:pur:roofing:1", (plant, {"amount": "35000.00"}))]}))
must(api("POST", f"/assets/{site}/settlements/{rest['settlement_date']}/{rest['seq']}/settle", ACCOUNTANT))
assert cost(plant, "01") == D("35000.00")
def test_the_auc_itself_never_depreciates_and_a_settlement_can_be_reversed_until_depreciated(auc):
site, build, plant = auc
v = next(x for x in must(api("GET", f"/assets/{site}", ACCOUNTANT))["values"] if x["area_code"] == "01")
assert v["planned_to_date"] == "0.00" # key NONE
d = must(api("POST", f"/assets/{site}/settlements", ACCOUNTANT, {"rules": [rule("acq:pur:steel:1", (build, {"percent": "100"}))]}))
must(api("POST", f"/assets/{site}/settlements/{d['settlement_date']}/{d['seq']}/settle", ACCOUNTANT))
r = must(api("POST", f"/assets/{site}/settlements/{d['settlement_date']}/{d['seq']}/reverse", ACCOUNTANT,
{"reason": "wrong receiver"}))
assert r["status"] == "reversed" and cost(build, "01") == D("0.00") and cost(site, "01") == D("225000.00")
def test_two_writers_taking_the_same_line_item_at_once_collide_on_its_claim(engine):
from sqlalchemy import insert
from sqlalchemy.exc import IntegrityError
from brick_fixed_assets.infrastructure import tables as T
row = {"auc_asset_id": "A", "line_item_transaction_id": "L", "n": 1, "settlement_seq": 1}
with engine.begin() as conn:
conn.execute(insert(T.fa_settlement_line_seq).values(**row))
with pytest.raises(IntegrityError):
with engine.begin() as conn:
conn.execute(insert(T.fa_settlement_line_seq).values(**{**row, "settlement_seq": 2}))