dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
tests/test_migration.py
"""FA-30 (load the Odoo register read-only: twice changes nothing, takeovers post no journal) and FA-31 (the owner signs only
what agrees with Odoo or is explained; a signed run is frozen)."""
from decimal import Decimal as D
import pytest
from conftest import ACCOUNTANT, APPROVER, OWNER, SYSTEM, api, must, on_day, setup_policy
from sqlalchemy import select
from brick_fixed_assets.infrastructure import odoo
CUT = "2026-09-30"
RUN = f"odoo:erp:{CUT}"
LOAD = {"odoo_db": "erp", "cutover_date": CUT, "company_code": "UK01", "odoo_flavour": "enterprise", "asset_numbers": "keep",
"draft_assets": "recreate", "conventions": "keep_asset", "opening_pools": "entered", "analytic": "drop"}
class FakeOdoo:
"""account.asset / account.move / trial balance rows as Odoo's read-only API answers them."""
def __init__(self):
model = {"id": 1, "name": "Vehicles", "state": "model", "method": "linear", "method_number": 60, "method_period": 1,
"account_asset_id": [100, "Vehicles"], "account_depreciation_id": [101, "Acc dep"],
"account_depreciation_expense_id": [600, "Expense"], "prorata_computation_type": "none_with_month"}
def asset(i, name, cost, acq, state="open", **kw):
return {"id": i, "name": name, "state": state, "model_id": [1, "Vehicles"], "original_value": cost, "book_value": 0,
"salvage_value": 0, "acquisition_date": acq, "method": "linear", "method_number": 60, "method_period": 1,
"account_asset_id": [100, "Vehicles"], "account_depreciation_id": [101, "Acc dep"],
"account_depreciation_expense_id": [600, "Expense"], "already_depreciated_amount_import": 0,
"prorata_computation_type": "none_with_month", **kw}
self.rows = [model, asset(11, "VAN-1", 12000, "2025-10-01"), asset(12, "VAN-2", 6000, "2026-03-01"),
asset(13, "CAR-OLD", 9000, "2023-01-01", state="close", disposal_date="2026-05-10"),
asset(14, "CAR-ANCIENT", 5000, "2019-01-01", state="close", disposal_date="2024-02-01")]
self.moves_by = {11: [(f"2025-{m:02d}-28", 200) for m in (10, 11, 12)] + [(f"2026-{m:02d}-28", 200) for m in range(1, 10)],
12: [(f"2026-{m:02d}-28", 100) for m in range(3, 10)],
13: [(f"2023-{m:02d}-28", 150) for m in range(1, 13)] + [(f"2024-{m:02d}-28", 150) for m in range(1, 13)]
+ [(f"2025-{m:02d}-28", 150) for m in range(1, 13)] + [(f"2026-{m:02d}-28", 150) for m in range(1, 5)]}
for r in self.rows: # book value = cost - posted depreciation
if r["state"] in ("open", "close") and r["id"] in self.moves_by:
r["book_value"] = r["original_value"] - sum(v for _, v in self.moves_by[r["id"]])
for r in self.rows[3:]:
r["book_value"] = 0 # a disposed asset has no book value left
self.extra_journal = D("0") # a manual journal hit the cost account in Odoo
def assets(self, db):
return [dict(r) for r in self.rows]
def moves(self, db, asset_ids, cutover):
return [{"id": 1000 * a + i, "asset_id": [a, "x"], "date": d, "state": "posted", "depreciation_value": v}
for a in asset_ids for i, (d, v) in enumerate(self.moves_by.get(a, [])) if d <= cutover.isoformat()]
def trial_balance(self, db, account_id, cutover):
if account_id == 100:
return D("18000.00") + self.extra_journal # 12,000 + 6,000: the disposed ones are off the account
return -D(str(sum(v for a in (11, 12) for _, v in self.moves_by.get(a, [])))) # a credit balance
@pytest.fixture
def odoo_world(gl, monkeypatch):
setup_policy(classes=(), start="2023-01-01", ifrs=False, tax=False)
gl.map.update({("account.account", 100): 150, ("account.account", 101): 151, ("account.account", 600): 771})
fake = FakeOdoo()
monkeypatch.setattr(odoo, "READER", fake)
return fake
def load():
return api("POST", "/migration/odoo/load", SYSTEM, LOAD)
def the_run(engine):
from brick_fixed_assets.infrastructure.tables import fa_migration_run
with engine.connect() as c:
return dict(c.execute(select(fa_migration_run)).first()._mapping)
def snapshot(engine):
"""Every table the migration may touch, as comparable rows."""
from brick_fixed_assets.infrastructure.tables import md
out = {}
with engine.connect() as c:
for t in md.sorted_tables:
out[t.name] = sorted(str(sorted((k, str(v)) for k, v in r._mapping.items() if k not in ("created_at", "updated_at")))
for r in c.execute(select(t)))
return out
def reconcile():
return must(api("POST", f"/migration/odoo/{RUN}/reconcile", SYSTEM, {"company_code": "UK01"}))
def assets_of(run=True):
return {a["asset_id"]: a for a in must(api("GET", "/assets", ACCOUNTANT))["assets"]}
def values(asset_id):
return must(api("GET", f"/assets/{asset_id}", ACCOUNTANT))["values"][0]
# ── FA-30 ───────────────────────────────────────────────────────────────────────────────────
def test_the_load_carries_cost_accumulated_depreciation_and_the_years_charge_and_posts_no_journal(odoo_world, gl):
out = must(load())
assert out["stats"] == {"classes": 1, "assets": 3, "written": 3, "unchanged": 0, "takeovers": 3, "retirements": 1}
got = assets_of()
assert sorted(got) == ["odoo:erp:account.asset:11", "odoo:erp:account.asset:12", "odoo:erp:account.asset:13"] # 14 closed in 2024
v = values("odoo:erp:account.asset:11")
assert (v["cost"], v["accumulated_depreciation"], v["net_book_value"]) == ("12000.00", "2400.00", "9600.00") # 600 + 1,800
assert values("odoo:erp:account.asset:12")["accumulated_depreciation"] == "700.00"
assert values("odoo:erp:account.asset:13")["net_book_value"] == "0.00" and got["odoo:erp:account.asset:13"]["status"] == "retired"
assert got["odoo:erp:account.asset:11"]["asset_no"] == "VAN-1"
assert gl.journals == {} and gl.calls == [] # gl migrates the journals; the takeover posts nothing
tx = must(api("GET", "/asset-transactions", ACCOUNTANT, {"asset_id": "odoo:erp:account.asset:11"}))["transactions"]
assert [(t["kind"], t["status"], t["value_date"]) for t in tx] == [("takeover", "posted", CUT)]
# life left: 60 - 12 periods already depreciated = 48; October is the next planned period, charged HERE, never Odoo's September
with on_day("2026-10-31T09:00:00+00:00"):
r = must(api("POST", "/depreciation-runs", ACCOUNTANT, {"company_code": "UK01", "period_id": "2026-10"}))
assert r["totals"] == {"01": "300.00"} # 200.00 + 100.00: not 400 + 200 (nor September again)
def test_a_second_load_changes_nothing(odoo_world, engine, gl):
must(load())
before = snapshot(engine)
again = must(load())
assert again["stats"]["written"] == 0 and again["stats"]["unchanged"] == 3 and again["stats"]["takeovers"] == 0
assert snapshot(engine) == before # not a single row written, version or timestamp aside
assert gl.journals == {}
def test_a_changed_odoo_row_is_written_by_a_reload_and_only_that_row(odoo_world, engine):
must(load())
odoo_world.moves_by[12].append(("2026-09-29", 100)) # one more posted move in Odoo
again = must(api("POST", f"/migration/odoo/{RUN}/reload", SYSTEM, {"company_code": "UK01"}))
assert again["stats"]["written"] == 1 and again["stats"]["unchanged"] == 2
assert values("odoo:erp:account.asset:12")["accumulated_depreciation"] == "800.00"
def test_renumbered_assets_take_the_class_series_and_a_second_load_claims_no_new_number(odoo_world):
must(api("POST", "/migration/odoo/load", SYSTEM, {**LOAD, "asset_numbers": "renumber"}))
nums = sorted(a["asset_no"] for a in assets_of().values())
assert nums == ["2026-VEHICLES-0001", "2026-VEHICLES-0002", "2026-VEHICLES-0003"]
must(api("POST", "/migration/odoo/load", SYSTEM, {**LOAD, "asset_numbers": "renumber"}))
must(api("POST", "/migration/odoo/load", SYSTEM, {**LOAD, "asset_numbers": "renumber"}))
from conftest import create_asset
new = create_asset("after-migration", cls="VEHICLES", description="Bought after cutover")
assert new["asset_no"] == "2026-VEHICLES-0004" # above the high-water mark, nothing reused or re-claimed
def test_every_door_must_be_answered_and_a_missing_gl_account_stops_the_load(odoo_world, gl):
s, b = api("POST", "/migration/odoo/load", SYSTEM, {**LOAD, "analytic": "maybe"})
assert s in (409, 422), b
gl.map.pop(("account.account", 100))
s, b = load()
assert s == 409 and "accounts_mapped" in str(b), b
# ── FA-31 ───────────────────────────────────────────────────────────────────────────────────
def test_signing_is_refused_while_a_must_equal_check_differs_and_a_difference_needs_its_explanation(odoo_world):
must(load())
odoo_world.extra_journal = D("500.00") # a manual journal hit the cost account in Odoo
rec = reconcile()
cost = next(c for c in rec["checks"] if c["check"] == "cost per class account at cutover")
assert cost["equal"] is False and cost["differences"][0]["odoo"] == "18500.00"
s, b = api("POST", f"/migration/odoo/{RUN}/sign", OWNER)
assert s == 409 and "zero_unexplained" in str(b), b
assert api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "asset count by state", "reason": "x"})[0] == 409 # equal
must(api("POST", f"/migration/odoo/{RUN}/explain", APPROVER,
{"check": "cost per class account at cutover", "reason": "manual journal MISC/2026/0042 of 500.00 in Odoo"}))
assert api("POST", f"/migration/odoo/{RUN}/sign", ACCOUNTANT)[0] == 403 # only fa:migration:sign signs
signed = must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))
assert signed["status"] == "signed" and signed["signed_by_email"] == "owner@uk.example" and signed["signed_on"] == "2026-11-20"
assert signed["summary"]["checks"] # the run's checksum is recorded
def test_a_must_equal_difference_cannot_be_explained_away(odoo_world):
must(load())
odoo_world.rows[1]["book_value"] = 9000 # Odoo says 9,000.00; the takeover gives 9,600.00
rec = reconcile()
nbv = next(c for c in rec["checks"] if c["check"] == "NBV per asset at cutover")
assert nbv["equal"] is False and nbv["differences"][0]["asset_id"] == "odoo:erp:account.asset:11"
s, b = api("POST", f"/migration/odoo/{RUN}/explain", APPROVER, {"check": "NBV per asset at cutover", "reason": "trust me"})
assert s == 409 and "must_equal" in str(b)
s, b = api("POST", f"/migration/odoo/{RUN}/sign", OWNER)
assert s == 409 and "must_equal" in str(b), b
def test_a_signed_run_is_frozen_a_load_writes_nothing_and_an_unreconciled_run_cannot_be_signed(odoo_world, engine):
must(load())
assert api("POST", f"/migration/odoo/{RUN}/sign", OWNER)[0] == 409 # not reconciled yet
rec = reconcile()
assert all(c["equal"] in (True, None) for c in rec["checks"]), rec["checks"]
must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))
before = snapshot(engine)
odoo_world.moves_by[11].append(("2026-09-29", 999)) # Odoo changed after the signature
for s, b in (load(), api("POST", f"/migration/odoo/{RUN}/reload", SYSTEM, {"company_code": "UK01"}),
api("POST", f"/migration/odoo/{RUN}/reconcile", SYSTEM, {"company_code": "UK01"})):
assert s == 409 and "signed" in str(b), (s, b)
assert snapshot(engine) == before # signed history is never rewritten
assert values("odoo:erp:account.asset:11")["accumulated_depreciation"] == "2400.00"
assert api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "oops"})[0] == 409 # nor discarded