dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
tests/test_register.py
"""FA-03, FA-04, FA-05: the register — assets with their sub-numbers, dated changes, the AI suggestion."""
import pytest
from conftest import ACCOUNTANT, APPROVER, CUSTODIAN, SYSTEM, acquire, api, create_asset, must, person, setup_policy
from brick_fixed_assets.infrastructure import ai_steps
from brick_fixed_assets.infrastructure.tables import Repository
@pytest.fixture(autouse=True)
def policy(gl):
setup_policy(classes=("VEH", "PLANT"))
def test_an_asset_is_numbered_once_from_the_class_series_and_its_sub_numbers_have_their_own_life():
press = create_asset("press-1", cls="PLANT", description="Hydraulic press (frame)", serial_no="SN-9", inventory_no="TAG-0007",
cost_centre="SITE-A", location_id="HALL-1", custodian_email="Pat@uk.example")
assert press["asset_id"] == "UK01-2026-PLANT-0001-0" and press["asset_no"] == "2026-PLANT-0001" and press["sub_no"] == 0
assert (press["serial_no"], press["inventory_no"], press["cost_centre"], press["location_id"]) == (
"SN-9", "TAG-0007", "SITE-A", "HALL-1")
assert press["custodian_email"] == "pat@uk.example" and press["status"] == "planned" and press["capitalised_on"] is None
unit = must(api("POST", f"/assets/{press['asset_id']}/sub-assets", ACCOUNTANT, {
"request_id": "press-2", "description": "Hydraulic unit",
"areas": [{"area_code": "01", "key_code": "LIN4"}, {"area_code": "30", "key_code": "LIN4"}]}))
assert unit["asset_id"] == "UK01-2026-PLANT-0001-1" and unit["asset_no"] == "2026-PLANT-0001" and unit["sub_no"] == 1
lives = {a["area_code"]: (a["key_code"], a["life_periods"]) for a in unit["areas"]}
assert lives == {"01": ("LIN4", 48), "30": ("LIN4", 48), "CA": ("LIN5", 60)} # overridable per area
frame = {a["area_code"]: (a["key_code"], a["life_periods"]) for a in press["areas"]}
assert frame == {"01": ("LIN5", 60), "30": ("LIN5", 60), "CA": ("LIN5", 60)} # copied from the class
assert unit["cost_centre"] == "SITE-A" and unit["status"] == "planned" # nothing capitalised yet
assert [x["asset_id"] for x in must(api("GET", "/assets", ACCOUNTANT))["assets"]] == [press["asset_id"], unit["asset_id"]]
def test_the_same_request_id_returns_the_same_asset_and_burns_no_number():
first = create_asset("same")
again = create_asset("same")
assert again["asset_id"] == first["asset_id"] and again["replay"] is True and first["replay"] is False
assert create_asset("next")["asset_no"] == "2026-VEH-0002"
sub = {"request_id": "sub-same", "description": "Tow bar"}
a = must(api("POST", f"/assets/{first['asset_id']}/sub-assets", ACCOUNTANT, sub))
b = must(api("POST", f"/assets/{first['asset_id']}/sub-assets", ACCOUNTANT, sub))
assert a["asset_id"] == b["asset_id"] and b["replay"] is True
def test_numbers_are_never_reused_and_gaps_are_not_filled(engine):
made = [create_asset(f"r{i}") for i in range(3)]
gone = made[1]
must(api("POST", f"/assets/{gone['asset_id']}/deactivate", ACCOUNTANT, {"reason": "ordered twice"}))
assert create_asset("r3")["asset_no"] == "2026-VEH-0004"
again = must(api("GET", f"/assets/{gone['asset_id']}", ACCOUNTANT))
assert again["status"] == "deactivated" and again["asset_no"] == "2026-VEH-0002" # kept with its number
def test_two_writers_claiming_the_same_number_collide_and_the_loser_takes_the_next(engine):
"""The claim is a plain INSERT keyed on the rendered number: a writer that read the same high-water mark collides."""
from sqlalchemy.exc import IntegrityError
with engine.begin() as conn:
repo = Repository(conn)
first = repo.claim_number("VEH", 2026, "{yyyy}-{class}-{seq:4}", "a")
assert first == "2026-VEH-0001"
with pytest.raises(IntegrityError):
with engine.begin() as conn: # a second transaction whose snapshot predates the first claim
Repository(conn).claim_number("VEH", 2026, "{yyyy}-{class}-{seq:4}", "b", seen_max=0)
with engine.begin() as conn:
assert Repository(conn).claim_number("VEH", 2026, "{yyyy}-{class}-{seq:4}", "b") == "2026-VEH-0002"
assert Repository(conn).number_gaps("VEH", 2026) == []
def test_an_asset_with_transactions_is_retired_not_deactivated():
a = create_asset("x")
s, b = acquire(a["asset_id"])
assert s == 200
s, b = api("POST", f"/assets/{a['asset_id']}/deactivate", ACCOUNTANT, {"reason": "oops"})
assert s == 409 and "retire it instead" in b["detail"]
def test_a_class_with_no_row_in_force_refuses_the_asset():
s, b = api("POST", "/assets", ACCOUNTANT, {"request_id": "r", "company_code": "UK01", "class_code": "GHOST",
"description": "x"})
assert s == 409 and "class_in_force" in b["detail"]
# ── FA-04 ────────────────────────────────────────────────────────────────────────────────────
def test_a_dated_change_of_cost_centre_and_life_starts_after_today_and_keeps_history():
a = create_asset("van", cost_centre="SITE-A")
s, b = api("POST", f"/assets/{a['asset_id']}/master", ACCOUNTANT, {"valid_from": "2026-11-20", "cost_centre": "SITE-B"})
assert s == 409 and "starts_after_today" in b["detail"]
s, b = api("POST", f"/assets/{a['asset_id']}/master", ACCOUNTANT, {"valid_from": "2026-11-19", "cost_centre": "SITE-B"})
assert s == 409
got = must(api("POST", f"/assets/{a['asset_id']}/master", ACCOUNTANT, {
"valid_from": "2026-12-01", "cost_centre": "SITE-B", "areas": [{"area_code": "01", "life_periods": 72}]}))
assert got["cost_centre"] == "SITE-A" # still SITE-A today: the change is dated
assert [m["cost_centre"] for m in got["master"]] == ["SITE-A", "SITE-B"]
one = [r for r in got["areas"] if r["area_code"] == "01"]
assert [(r["valid_from"], r["life_periods"]) for r in one] == [("1900-01-01", 60), ("2026-12-01", 72)]
assert api("POST", f"/assets/{a['asset_id']}/master", ACCOUNTANT, {"valid_from": "2026-12-01"})[0] == 409 # says nothing
def test_a_stale_version_is_a_409_and_a_mass_change_lists_each_asset_it_touched():
a = create_asset("v1")
b = create_asset("v2")
must(api("POST", f"/assets/{a['asset_id']}/master", ACCOUNTANT, {"valid_from": "2026-12-01", "cost_centre": "X"}))
s, body = api("POST", f"/assets/{a['asset_id']}/master", ACCOUNTANT, {"valid_from": "2026-12-02", "cost_centre": "Y",
"version": a["version"]})
assert s == 409 and "changed since you opened it" in body["detail"]
mass = must(api("POST", "/asset-classes/VEH/mass-change", ACCOUNTANT, {
"valid_from": "2027-01-01", "areas": [{"area_code": "01", "life_periods": 72}]}))
assert mass["touched"] == [a["asset_id"], b["asset_id"]] and mass["skipped"] == []
bad = must(api("POST", "/asset-classes/VEH/mass-change", ACCOUNTANT, {"valid_from": "2026-11-20", "cost_centre": "Z"}))
assert bad["touched"] == [] and len(bad["skipped"]) == 2 and "starts after today" in bad["skipped"][0]["reason"]
# ── who sees which asset ─────────────────────────────────────────────────────────────────────
def test_a_custodian_sees_only_the_assets_they_look_after_and_another_sites_asset_is_not_found():
mine = create_asset("m", cost_centre="SITE-A", custodian_email="asset_custodian@uk.example")
other = create_asset("o", cost_centre="SITE-B", custodian_email="someone.else@uk.example")
assert [x["asset_id"] for x in must(api("GET", "/my/assets", CUSTODIAN))["assets"]] == [mine["asset_id"]]
assert must(api("GET", f"/my/assets/{mine['asset_id']}", CUSTODIAN))["asset_id"] == mine["asset_id"]
assert api("GET", f"/my/assets/{other['asset_id']}", CUSTODIAN)[0] == 404 # not found, not forbidden
assert api("GET", f"/assets/{mine['asset_id']}", CUSTODIAN) == (403, {"detail": "missing permission: fa:asset:read"})
nobody = person("", ["fa:asset:read_own"], sub="") # no identity at all: sees nothing
assert api("GET", "/my/assets", nobody)[0] == 401 # no identity at all: refused at the door
by_sub = person("other.login@uk.example", ["fa:asset:read_own"], sub="asset_custodian@uk.example")
assert [x["asset_id"] for x in must(api("GET", "/my/assets", by_sub))["assets"]] == [mine["asset_id"]] # scoped by sub too
assert len(must(api("GET", "/assets", ACCOUNTANT))["assets"]) == 2 # the accountant sees every asset
# ── FA-05: the AI step ───────────────────────────────────────────────────────────────────────
def test_the_ai_step_only_suggests_and_nothing_applies_without_the_accountant(monkeypatch):
a = create_asset("van", description="Ford Transit 350 L3H2, CO2 210 g/km", cls="PLANT")
answer = ai_steps.Answer("ai_suggest_asset_class", "suggest", "done",
{"class_code": "VEH", "life_periods:01": 60, "ca_pool": "special_rate"},
"a van, not a car (payload over 1 tonne)", 88, "kimi-k2.6")
monkeypatch.setattr(ai_steps, "ask", lambda *a, **k: answer)
out = must(api("POST", f"/assets/{a['asset_id']}/suggest-class", SYSTEM, {}))
assert out["kind"] == "done" and out["baseline"]["class_code"] == "PLANT" and out["run_no"] == 1
shown = must(api("GET", f"/assets/{a['asset_id']}", ACCOUNTANT))
assert shown["class_code"] == "PLANT" and len(shown["suggestions"]) == 3 # suggested, applied to nothing
assert all(f["decision"] is None for f in shown["suggestions"])
by_field = {f["field"]: f["finding_no"] for f in shown["suggestions"]}
base = f"/assets/{a['asset_id']}/suggestions/1"
ok = must(api("POST", f"{base}/{by_field['class_code']}/decide", ACCOUNTANT, {"decision": "accepted"}))
assert ok["applied"] == "class_code=VEH"
fixed = must(api("POST", f"{base}/{by_field['ca_pool']}/decide", ACCOUNTANT, {"decision": "corrected", "value": "main"}))
assert fixed["applied"] == "ca_pool=main"
gone = must(api("POST", f"{base}/{by_field['life_periods:01']}/decide", ACCOUNTANT, {"decision": "dismissed"}))
assert gone["applied"] is None
after = must(api("GET", f"/assets/{a['asset_id']}", ACCOUNTANT))
assert after["class_code"] == "VEH" and after["ca_pool"] == "main"
assert sorted(f["decision"] for f in after["suggestions"]) == ["accepted", "corrected", "dismissed"]
assert all(f["decided_by_email"] == "asset_accountant@uk.example" and f["model"] == "kimi-k2.6" for f in after["suggestions"])
s, b = api("POST", f"{base}/{by_field['ca_pool']}/decide", ACCOUNTANT, {"decision": "dismissed"})
assert s == 409 and "already corrected" in b["detail"]
def test_with_no_provider_the_step_leaves_a_task_and_off_runs_nothing(engine):
a = create_asset("v")
out = must(api("POST", f"/assets/{a['asset_id']}/suggest-class", SYSTEM, {}))
assert out["kind"] == "manual" and out["baseline"]["areas"] and out["task_id"] # the class defaults are the fallback
assert api("POST", f"/assets/{a['asset_id']}/suggest-class", ACCOUNTANT, {})[0] == 403 # a machine permission
must(api("POST", "/settings/day-to-day", APPROVER, {"setting_key": "ai.ai_suggest_asset_class.mode", "value": "off"}))
assert must(api("POST", f"/assets/{a['asset_id']}/suggest-class", SYSTEM, {}))["kind"] == "off"