dodil/erp-brick-gl
DODIL ERP brick: general ledger
tests/test_migration.py
"""GL-34 / GL-35: the Odoo books carried, reconciled per account and period to the penny, and signed by the owner."""
import json
from datetime import date
from decimal import Decimal
import fake_odoo
import pytest
from conftest import ACCOUNTANT, AUDITOR, BOOKKEEPER, OWNER, SYSTEM, api, must, person, refused
from erp_platform.seam import derived_id
LOAD = person("migration@svc.example", ["gl:migration:load", "gl:migration:reconcile", "gl:migration:read"])
def doors(cutover="2026-09-30", **kw):
return {"db": "acme", "cutover_date": cutover, "companies": [{"odoo_id": 1, "company_code": "UK01", "country": "GB"}], "ledger_id": "IFRS",
"accounting_principle": "IFRS", "history": "two_years", "profit_centre_plan": "Profit", "cost_centre_plan": "Dept", "segments": {"*": "ALL"}, **kw}
@pytest.fixture
def odoo_world():
f = fake_odoo.install()
yield f
fake_odoo.uninstall()
def load(d=None, who=LOAD):
return api("POST", "/migration/odoo/load", who, {"doors": d or doors()})
def test_the_gates_partners_and_taxes_must_be_mapped_first():
f = fake_odoo.install(partners={500: 5000, 700: 7000}, taxes={900: "S"})
refused(load(), 409, "[partners_mapped]", "701", "business-partner migration")
fake_odoo.install(taxes={})
refused(load(), 409, "[taxes_mapped]", "tax migration")
fake_odoo.uninstall()
_ = f
def test_the_load_writes_history_keyed_on_the_odoo_record_and_a_second_run_changes_nothing(odoo_world):
out = must(load())
c = out["counts"]
assert out["run_id"] == "odoo:acme:2026-09-30" and out["status"] == "loaded"
assert c["company"]["created"] == 1 and c["account"]["created"] == 8 and c["journal"]["created"] == 8 # opening + 7 moves (2025 and 2026 to Sept)
assert c["profit_centre"]["created"] == 2 and c["cost_centre"]["created"] == 1 and c["fx_rate"]["created"] == 2
again = must(load())
assert all(v["created"] == 0 for v in again["counts"].values() if "created" in v), again["counts"]
# the journal is Odoo's move, to the penny
jid = derived_id("odoo:acme:account.move:101")
j = must(api("GET", f"/journals/{jid}", ACCOUNTANT))
assert j["source_module"] == "odoo" and j["source_document"] == "odoo:acme:account.move:101" and j["book"] == "sales" and j["entry_no"] == 1
assert j["number"] == "SALES/FY2025/0001" and [(ln["debit"], ln["credit"]) for ln in j["lines"]] == [("1200.00", "0.00"), ("0.00", "1000.00"), ("0.00", "200.00")]
assert [ln["odoo_line_id"] for ln in j["lines"]] == ["1010", "1011", "1012"] and [ln["bp_id"] for ln in j["lines"]] == ["7000", None, None]
assert j["lines"][1]["profit_centre"] == "PC-LON" and j["lines"][2]["tax_code"] == "S"
# a move split across two profit centres becomes split lines in proportion, to the penny
j4 = must(api("GET", f"/journals/{derived_id('odoo:acme:account.move:104')}", ACCOUNTANT))
rev = [s for s in j4["split_lines"] if s["account_id"] == str(derived_id("odoo:acme:account.account:6"))]
assert sorted((s["profit_centre"], s["credit"]) for s in rev) == [("PC-LON", "300.00"), ("PC-WHS", "200.00")]
# the foreign line keeps its own currency
j5 = must(api("GET", f"/journals/{derived_id('odoo:acme:account.move:105')}", ACCOUNTANT))
assert j5["currency"] == "USD" and j5["lines"][0]["currency"] == "USD" and j5["lines"][0]["amount_currency"] == "1000.00" and j5["lines"][0]["functional_debit"] == "800.00"
# the reversal links back; the original reads reversed
j6 = must(api("GET", f"/journals/{derived_id('odoo:acme:account.move:106')}", ACCOUNTANT))
assert j6["reverses_journal_id"] == str(derived_id("odoo:acme:account.move:103"))
assert must(api("GET", f"/journals/{derived_id('odoo:acme:account.move:103')}", ACCOUNTANT))["status"] == "reversed"
# periods are closed up to Odoo's lock date
ps = {p["period_id"]: p["status"] for p in must(api("GET", "/periods", ACCOUNTANT, {"fiscal_year": "FY2026"}))["periods"]}
assert ps["2026-08"] == "closed" and ps["2026-09"] == "open"
assert must(api("GET", "/audit/verify-chain", AUDITOR, {"company_code": "UK01"}))["met"] is True
def test_the_id_map_route_answers_what_was_migrated_and_omits_the_rest(odoo_world):
must(load())
rows = must(api("GET", "/migration-map/acme/account.account", LOAD, {"ids": "1,2,99"}))
assert [(r["odoo_model"], r["odoo_id"]) for r in rows] == [("account.account", "1"), ("account.account", "2")]
assert rows[0]["ours"] == str(derived_id("odoo:acme:account.account:1"))
assert must(api("GET", "/migration-map/acme/account.journal", LOAD, {"ids": "10,11,12,13"}))[0]["ours"] == "sales"
assert must(api("GET", "/migration-map/acme/res.company", LOAD, {"ids": "1"}))[0]["ours"] == "UK01"
assert must(api("GET", "/migration-map/acme/account.account", LOAD, {"ids": "99"})) == []
assert api("GET", "/migration-map/acme/account.account", BOOKKEEPER, {"ids": "1"})[0] == 403
def test_the_numbers_continue_after_the_highest_migrated_one(odoo_world):
from world import account_id, line, post
must(load())
from conftest import POSTER
s, b = post("INV-N", "issue", "2026-10-05", [line(purpose="receivable", debit="10.00"), line(purpose="revenue", credit="10.00")], who=POSTER)
assert s == 200 and b["number"] == "SALES/FY2026/0003", b # INV/2026/0001 and 0002 were migrated
_ = account_id
def reconcile(run_id="odoo:acme:2026-09-30"):
return must(api("POST", "/migration/odoo/reconcile", LOAD, {"run_id": run_id}))
def checks_of(run):
return {c["check"]: c for c in run["checks"]}
def test_the_trial_balance_per_account_and_period_equals_odoos_to_the_penny(odoo_world):
must(load())
r = reconcile()
by = checks_of(r)
for name, c in by.items():
if c["must"] == "equal":
assert c["equal"] is True, (name, c["odoo"], c["ours"])
assert by["trial balance at cutover nets to zero on both sides"]["ours"] == "0.00"
assert by["draft moves dated on or before cutover"]["equal"] is False and by["draft moves dated on or before cutover"]["must"] == "explained"
assert by["receivable control account against payments' open balance at cutover"]["equal"] is False
def test_signing_is_refused_until_the_explainable_checks_are_explained_and_the_owner_alone_signs(odoo_world):
must(load())
rid = reconcile()["run_id"]
refused(api("POST", "/migration/odoo/sign", OWNER, {"run_id": rid}), 409, "[zero_unexplained]", "draft moves")
drafts, receivable = "draft moves dated on or before cutover", "receivable control account against payments' open balance at cutover"
refused(api("POST", "/migration/odoo/explain-difference", ACCOUNTANT, {"run_id": rid, "check": "trial balance at cutover nets to zero on both sides", "reason": "x"}),
409, "nothing to explain")
must(api("POST", "/migration/odoo/explain-difference", ACCOUNTANT, {"run_id": rid, "check": drafts, "reason": "one draft invoice, re-entered by hand"}))
must(api("POST", "/migration/odoo/explain-difference", ACCOUNTANT, {"run_id": rid, "check": receivable, "reason": "payments has not migrated yet"}))
assert api("POST", "/migration/odoo/sign", ACCOUNTANT, {"run_id": rid})[0] == 403
signed = must(api("POST", "/migration/odoo/sign", OWNER, {"run_id": rid}))
assert signed["status"] == "signed" and signed["signed_by"] == "owner@firm.example" and signed["checksums"]
# explanations on unchanged checks are kept by a re-reconcile of an unsigned run; a signed run is frozen
refused(api("POST", "/migration/odoo/reconcile", LOAD, {"run_id": rid}), 409, "signed")
refused(load(), 409, "signed run is never rewritten")
def test_a_difference_in_the_trial_balance_cannot_be_explained_away(odoo_world):
f = odoo_world
must(load())
f.mlines[107][0]["debit"] = Decimal("121") # Odoo changed after the load
rid = reconcile()["run_id"]
tbk = checks_of(must(api("GET", f"/migration/odoo/runs/{rid}", LOAD)))["trial balance per account and period (debit, credit, closing)"]
assert tbk["equal"] is False and tbk["must"] == "equal"
refused(api("POST", "/migration/odoo/explain-difference", ACCOUNTANT, {"run_id": rid, "check": tbk["check"], "reason": "trust me"}), 409, "[must_equal]")
refused(api("POST", "/migration/odoo/sign", OWNER, {"run_id": rid}), 409, "[must_equal]")
f.mlines[107][0]["debit"] = Decimal("120")
again = must(api("POST", "/migration/odoo/reconcile", LOAD, {"run_id": rid}))
assert checks_of(again)[tbk["check"]]["equal"] is True
def test_explanations_on_unchanged_checks_are_kept_when_reloaded(odoo_world):
must(load())
rid = reconcile()["run_id"]
must(api("POST", "/migration/odoo/explain-difference", ACCOUNTANT, {"run_id": rid, "check": "draft moves dated on or before cutover", "reason": "re-entered"}))
out = must(api("POST", "/migration/odoo/reload", LOAD, {"doors": doors()}))
assert all(v["created"] == 0 for v in out["counts"].values() if "created" in v)
again = reconcile()
assert checks_of(again)["draft moves dated on or before cutover"]["explanation"] == "re-entered"
def test_a_later_run_only_adds_what_odoo_posted_after_the_signed_cutover(odoo_world):
must(load())
rid = reconcile()["run_id"]
for chk, why in (("draft moves dated on or before cutover", "re-entered"), ("receivable control account against payments' open balance at cutover", "later")):
must(api("POST", "/migration/odoo/explain-difference", ACCOUNTANT, {"run_id": rid, "check": chk, "reason": why}))
must(api("POST", "/migration/odoo/sign", OWNER, {"run_id": rid}))
before = must(api("GET", "/journals", ACCOUNTANT, {"limit": 500}))["journals"]
out = must(load(doors("2026-10-31")))
assert out["run_id"] == "odoo:acme:2026-10-31" and out["counts"]["journal"]["created"] == 1
after = must(api("GET", "/journals", ACCOUNTANT, {"limit": 500}))["journals"]
assert len(after) == len(before) + 1
new = [j for j in after if j["journal_id"] not in {b["journal_id"] for b in before}][0]
assert new["source_document"] == "odoo:acme:account.move:108" and new["journal_date"] == "2026-10-12"
for b in before: # nothing signed was rewritten
assert next(a for a in after if a["journal_id"] == b["journal_id"])["row_hash"] == b["row_hash"]
def test_an_unsigned_run_can_be_discarded_by_the_owner_and_loaded_again(odoo_world):
must(load())
rid = reconcile()["run_id"]
assert api("POST", "/migration/odoo/discard", ACCOUNTANT, {"run_id": rid, "reason": "wrong door"})[0] == 403
out = must(api("POST", "/migration/odoo/discard", OWNER, {"run_id": rid, "reason": "wrong history door"}))
assert out["status"] == "discarded" and out["journals_removed"] == 8
assert must(api("GET", "/journals", ACCOUNTANT, {"limit": 500}))["journals"] == []
refused(load(), 409, "discarded")
def test_recurring_deferrals_budgets_and_the_group_mapping_come_as_proposals(odoo_world):
f = odoo_world
f.recurring = [{"id": 900, "name": "Rent", "auto_post": "monthly", "auto_post_until": date(2027, 6, 30),
"lines": [{"account_id": 7, "debit": "2500"}, {"account_id": 4, "credit": "2500"}]}]
f.deferrals = [{"id": 77, "account_id": 5, "counter_account_id": 6, "deferred_end_date": date(2027, 3, 31), "balance": "1200", "released": "300"}]
f.budget_rows = [{"id": 5, "name": "B2026", "date_from": date(2026, 1, 1), "date_to": date(2026, 12, 31),
"lines": [{"account_id": 6, "analytic_account_id": 51, "planned_amount": "1000.00"}]}]
f.consolidation = [{"id": 1, "code": "G-4000", "name": "Group revenue", "type": "income", "account_ids": [6]}]
out = must(load())
c = out["counts"]
assert c["recurring"]["created"] == 1 and c["accrual"]["created"] == 1 and c["budget"]["created"] == 1 and c["group_mapping"]["created"] == 1
rec = must(api("GET", "/recurring", ACCOUNTANT, {}))["recurring"][0]
assert rec["recurring_id"] == "ODOO-900" and rec["status"] == "proposed" and rec["first_period"] == "2026-10" and rec["last_period"] == "2027-06"
acc = must(api("GET", "/accruals", ACCOUNTANT, {}))["accruals"][0]
assert acc["accrual_code"] == "ODOO-77" and acc["status"] == "proposed" and acc["amount"] == "900.00" and acc["start_period"] == "2026-10"
b = must(api("GET", "/budgets/ODOO-5", ACCOUNTANT, {}))
assert b["status"] == "approved" and len(b["lines"]) == 12 and sum(Decimal(x["amount"]) for x in b["lines"]) == Decimal("1000.00")
assert must(api("GET", f"/accounts/{derived_id('odoo:acme:account.account:6')}", ACCOUNTANT))["group_account_history"][0]["group_account_id"]
again = must(load())
assert all(v["created"] == 0 for v in again["counts"].values() if "created" in v)
def test_a_migration_is_a_machine_not_a_person(odoo_world):
for who in (ACCOUNTANT, OWNER, BOOKKEEPER):
assert api("POST", "/migration/odoo/load", who, {"doors": doors()})[0] == 403
assert api("POST", "/migration/odoo/load", LOAD, {"doors": {"db": "acme"}})[0] == 422
_ = (json, SYSTEM)