dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
tests/test_migration.py
"""The Odoo slice (AR-33, AR-34, SC-AR-08): history loaded read-only, reconciled to zero unexplained, signed by the owner.
A fake Odoo hands the use case rows as search_read returns them; there is no live Odoo anywhere."""
import json
from decimal import Decimal
from conftest import APPROVER, BILLER, BP, OWNER, SYSTEM, api, billing, hand_draft, issue, line, migrate_payments, must, person
from sqlalchemy import func, select
from brick_ar.infrastructure.tables import (
ar_credit_notes,
ar_invoices,
ar_migration_map,
ar_number_series,
)
LOADER = SYSTEM
RUN = "odoo:odooprod:2026-10-31"
INVOICES = [f"odoo:odooprod:{i}" for i in (100, 101, 102, 104)]
def load(**kw):
return api("POST", "/migration/odoo/load", LOADER, {"cutover_date": "2026-10-31", **kw})
def reconcile():
return must(api("POST", f"/migration/runs/{RUN}/reconcile", SYSTEM, {}))
def counts(engine):
with engine.begin() as c:
return {"invoices": c.execute(select(func.count()).select_from(ar_invoices)).scalar(),
"credits": c.execute(select(func.count()).select_from(ar_credit_notes)).scalar(),
"maps": c.execute(select(func.count()).select_from(ar_migration_map)).scalar()}
def checks_of(run):
return {(c["check"], c["slice"]): c for c in run["checks"]}
def test_ar_34_history_is_carried_exactly_as_odoo_stored_it(odoo, engine):
"""1,200 posted invoices and 40 refunds in the spec; here 4 invoices (one cancelled) and 1 refund: one invoice per move keyed
odoo:<db>:<id>, Odoo's number, dates, lines, tax and totals to the cent; credit notes linked to the invoice they reverse."""
run = must(load())
assert run["run_id"] == RUN and run["status"] == "loaded" and run["counts"]["invoices"] == 4 and run["counts"]["credit_notes"] == 1
inv = must(api("GET", "/invoices/odoo:odooprod:100", BILLER))
assert (inv["lifecycle"], inv["invoice_no"], inv["issue_date"], inv["tax_point_date"], inv["due_date"], inv["source_kind"], inv["bp_id"]) == (
"issued", "INV/2026/00001", "2026-03-01", "2026-03-01", "2026-03-31", "odoo", "7311")
assert (inv["subtotal"], inv["tax_total"], inv["total"], inv["terms_id"], inv["customer_po_ref"], inv["migration_run_id"]) == (
"1000.00", "200.00", "1200.00", "odoo:1", "PO-100", RUN)
assert inv["lines"][0]["odoo_line_id"] == "1001" and inv["lines"][0]["net_amount"] == "1000.00" and inv["lines"][0]["tax_code"] == "S"
assert inv["tax"][0]["tax_code"] == "S" and inv["tax"][0]["rounding_method"] == "odoo" and json.loads(inv["bill_to_json"])["bill_to_name"] == "Acme Ltd"
odd = must(api("GET", "/invoices/odoo:odooprod:102", BILLER))
assert odd["tax_total"] == "19.99" and odd["total"] == "119.99" # never recomputed to 20.00
eur = must(api("GET", "/invoices/odoo:odooprod:101", BILLER))
assert (eur["currency"], eur["fx_rate"], eur["total_functional"], eur["bp_id"]) == ("EUR", "0.8600", "430.00", "7312")
void = must(api("GET", "/invoices/odoo:odooprod:104", BILLER))
assert void["lifecycle"] == "void" and void["invoice_no"] == "INV/2026/00005" and void["void_reason"] == "cancelled in Odoo"
cn = must(api("GET", "/credit-notes/odoo:odooprod:103", BILLER))
assert (cn["status"], cn["credit_no"], cn["invoice_id"], cn["total"], cn["source_kind"]) == (
"approved", "RINV/2026/00001", "odoo:odooprod:100", "120.00", "odoo")
ids = {r["invoice_id"] for r in must(api("GET", "/invoices", BILLER, {"limit": 100}))["invoices"]}
assert ids == set(INVOICES) # no draft, no 2018 move, no vendor bill
def test_ar_34_the_second_load_changes_nothing(odoo, engine):
must(load())
before = counts(engine)
versions = {i: must(api("GET", f"/invoices/{i}", BILLER))["version"] for i in INVOICES}
again = must(load())
assert again["counts"]["unchanged"] == 5 and again["counts"]["invoices"] == 0 and again["counts"]["credit_notes"] == 0 and again["counts"]["updated"] == 0
assert counts(engine) == before and {i: must(api("GET", f"/invoices/{i}", BILLER))["version"] for i in INVOICES} == versions
def test_ar_34_a_migrated_document_is_never_recomputed_reposted_resent_or_raised_again(odoo, world):
must(load())
assert world.tax.calls == [] and world.gl.journals == {}
iid = "odoo:odooprod:100"
s, b = api("POST", f"/invoices/{iid}/send", BILLER, {"address": "x@y.example"})
assert s == 409 and "history" in b["detail"]
s, b = api("POST", f"/invoices/{iid}/edit", BILLER, {"customer_po_ref": "x"})
assert s == 409 and "credit it instead" in b["detail"]
from conftest import account_map
account_map()
skipped = must(api("POST", f"/documents/{iid}/post", SYSTEM, {}))["results"][0]
assert skipped["posted"] is False and "never posted" in skipped["reason"] # the post path skips it
assert world.gl.journals == {}
assert must(api("POST", "/documents/post-pending", SYSTEM, {}))["posted"] == 0 and world.gl.journals == {}
assert must(api("POST", "/schedules/raise-due", SYSTEM, {}))["raised"] == []
world.remove("gl")
assert must(api("POST", "/exports", BILLER, {"period": "2026-03"}))["exported"] == 0 # exports are for what this brick issued
r = must(api("GET", f"/invoices/{iid}/readiness", BILLER))
assert "post_to_ledger" not in r.get("transitions", {}) and "export" not in r.get("transitions", {})
def test_ar_34_credit_notes_and_allocations_do_run_on_a_migrated_invoice(odoo, world):
must(load())
migrate_payments(world, odoo)
iid = "odoo:odooprod:100"
cn = must(api("POST", "/credit-notes", BILLER, {"invoice_id": iid, "reason": "late correction", "lines": [line("c", "1", "50.00")]}))
world.payments.credit_faces[cn["credit_id"]] = Decimal(cn["total"])
must(api("POST", f"/credit-notes/{cn['credit_id']}/approve", APPROVER, {}))
assert must(api("GET", f"/invoices/{iid}", BILLER))["settlement"]["outstanding"] == "640.00"
def test_the_load_waits_for_the_partner_and_tax_migrations(odoo, world):
world.bp.migration_map.pop(11)
world.tax.migration_map.pop(2)
s, b = load()
assert s == 409 and "partners_mapped" in b["detail"] and "[11]" in b["detail"] and "run the business-partner migration first" in b["detail"]
assert "taxes_mapped" in b["detail"] and "[2]" in b["detail"] and "run the tax migration first" in b["detail"]
assert api("GET", "/invoices/odoo:odooprod:100", BILLER)[0] == 404 # nothing was written
world.remove("tax")
s, b = load()
assert s == 409 and "tax is not answering" in b["detail"]
s, b = load(doors={"numbering": "sometimes"})
assert s == 409 and "doors_decided" in b["detail"]
def test_the_signed_decisions_are_the_default_doors(odoo):
run = must(load())
assert run["doors"] == {"history_years": 6, "numbering": "continue", "carry_drafts": False, "instalments": "as_odoo", "company_id": None, "ledger": None}
run2 = must(load(doors={"history_years": 20}))
assert run2["counts"]["invoices"] == 1 and api("GET", "/invoices/odoo:odooprod:106", BILLER)[0] == 200 # the 2018 move: within 20 years
def test_odoo_drafts_are_not_carried_unless_the_door_says_so(odoo):
must(load())
assert api("GET", "/invoices/odoo:odooprod:105", BILLER)[0] == 404
must(load(doors={"carry_drafts": True}))
assert must(api("GET", "/invoices/odoo:odooprod:105", BILLER))["lifecycle"] == "draft"
def test_odoos_numbering_continues_after_the_cutover(odoo, world, engine):
"""Continue Odoo's invoice numbering: the next invoice is INV/2026/00006, in the series Odoo's last number was in."""
must(load())
with engine.begin() as c:
assert c.execute(select(ar_number_series.c.high_water_mark).where(ar_number_series.c.series_id == "INV/2026")).scalar() == 5
migrate_payments(world, odoo)
reconcile()
explain_and_sign(world)
must(__import__("conftest").control("invoice_number_format", "INV/{yyyy}/{seq:5}"))
from conftest import on_day
with on_day("2026-11-21"):
billing()
assert issue(hand_draft()["invoice_id"], world=world)["invoice_no"] == "INV/2026/00006"
def test_the_customers_billing_profiles_and_terms_come_over(odoo):
must(load())
got = must(api("GET", f"/customers/{BP}/billing", BILLER))["in_force"]
assert (got["bill_to_name"], got["terms_id"], got["valid_from"], got["invoice_email"]) == ("Acme Ltd", "odoo:1", "2026-10-31", "ap@acme.example")
assert {"terms_id": "odoo:1", "days": 30, "basis": "invoice_date"}.items() <= next(t for t in must(api("GET", "/terms", BILLER))["terms"] if t["terms_id"] == "odoo:1").items()
# ── the reconciliation and the signature ────────────────────────────────────────────────────────
def explain_and_sign(world):
run = must(api("GET", f"/migration/runs/{RUN}", OWNER))
for c in run["checks"]:
if not c["equal"] and c["must"] == "explained" and not c["explanation"]:
must(api("POST", f"/migration/runs/{RUN}/explain", APPROVER, {"check": c["check"], "slice": c["slice"], "reason": "Odoo test invoice deleted before go-live"}))
return must(api("POST", f"/migration/runs/{RUN}/sign", OWNER, {}))
def test_sc_ar_08_loaded_twice_reconciled_one_difference_explained_and_signed(odoo, world):
"""Posted invoices and refunds, partly paid, partners and taxes mapped: counts, totals and open balance per customer equal
Odoo's; the one explained difference is shown with its reason; signing is refused before the explanation and accepted after."""
must(load())
must(load())
migrate_payments(world, odoo)
run = reconcile()
by = checks_of(run)
assert run["status"] == "reconciled"
assert by[("document_count", "out_invoice|posted|2026")]["odoo"] == "3" and by[("document_count", "out_invoice|cancel|2026")]["equal"]
assert by[("total_amount", "out_invoice|GBP|2026")]["odoo"] == "1379.99" == by[("total_amount", "out_invoice|GBP|2026")]["ours"]
assert by[("total_tax", "out_refund|GBP|2026")]["equal"] and by[("tax_amount", "S|2026-Q1")]["ours"] == "200.00"
assert by[("open_balance", f"{BP}|GBP")]["odoo"] == "700.00" == by[("open_balance", f"{BP}|GBP")]["ours"]
assert by[("open_balance", "7312|EUR")]["equal"] and by[("aged_open_balance", f"{BP}|GBP|90+")]["equal"]
assert by[("receivable_balance", "GBP")]["equal"] and by[("lines_sum_to_document", "consistent")]["equal"]
gap = by[("numbering_gaps", "INV/2026")]
assert (gap["odoo"], gap["equal"], gap["must"]) == ("3", False, "explained")
assert all(c["equal"] for c in run["checks"] if c["must"] == "equal")
assert [g["gate"] for g in run["readiness"]["gates"] if not g["met"]] == ["zero_unexplained"]
s, b = api("POST", f"/migration/runs/{RUN}/sign", OWNER, {})
assert s == 409 and "zero_unexplained" in b["detail"] and "numbering_gaps / INV/2026" in b["detail"] and "explain each" in b["detail"]
explained = must(api("POST", f"/migration/runs/{RUN}/explain", APPROVER, {"check": "numbering_gaps", "slice": "INV/2026", "reason": "Odoo test invoice deleted before go-live"}))
row = checks_of(explained)[("numbering_gaps", "INV/2026")]
assert row["explanation"] == "Odoo test invoice deleted before go-live" and row["explained_by"] == "approver@shop.example"
signed = must(api("POST", f"/migration/runs/{RUN}/sign", OWNER, {}))
assert signed["status"] == "signed" and signed["signed_by"] == "owner@shop.example" and signed["signed_at"] and len(signed["checksums_json"]) == 64
assert must(api("POST", f"/migration/runs/{RUN}/sign", OWNER, {}))["checksums_json"] == signed["checksums_json"]
def test_ar_33_a_check_that_must_be_equal_cannot_be_explained_away(odoo, world):
must(load())
migrate_payments(world, odoo)
odoo.move_rows[0]["amount_total"] = 1300.0 # Odoo's figure moves after the load
run = reconcile()
bad = checks_of(run)[("total_amount", "out_invoice|GBP|2026")]
assert bad["equal"] is False and bad["difference"] == "-100.00" and bad["must"] == "equal"
s, b = api("POST", f"/migration/runs/{RUN}/explain", APPROVER, {"check": "total_amount", "slice": "out_invoice|GBP|2026", "reason": "because"})
assert s == 409 and "cannot be explained away" in b["detail"]
s, b = api("POST", f"/migration/runs/{RUN}/sign", OWNER, {})
assert s == 409 and "must_equal" in b["detail"] and "fix the mapping and reload" in b["detail"]
odoo.move_rows[0]["amount_total"] = 1200.0 # fixed at the source: reload, reconcile, sign
must(load())
run = reconcile()
assert all(c["equal"] for c in run["checks"] if c["must"] == "equal")
assert explain_and_sign(world)["status"] == "signed"
def test_the_open_balance_cannot_pass_before_payments_has_migrated(odoo, world):
must(load())
run = reconcile() # payments knows none of these invoices yet
by = checks_of(run)
assert by[("open_balance", f"{BP}|GBP")]["equal"] is False and by[("open_balance", f"{BP}|GBP")]["ours"] == "0.00"
s, b = api("POST", f"/migration/runs/{RUN}/sign", OWNER, {})
assert s == 409 and "must_equal" in b["detail"] and "open_balance" in b["detail"]
world.remove("payments")
run = reconcile()
assert checks_of(run)[("open_balance", f"{BP}|GBP")]["equal"] is False
world.install("payments")
migrate_payments(world, odoo)
assert all(c["equal"] for c in reconcile()["checks"] if c["must"] == "equal")
def test_a_reload_keeps_the_explanations_of_unchanged_checks_and_a_signed_run_is_final(odoo, world):
must(load())
migrate_payments(world, odoo)
reconcile()
must(api("POST", f"/migration/runs/{RUN}/explain", APPROVER, {"check": "numbering_gaps", "slice": "INV/2026", "reason": "deleted test invoice"}))
must(load())
again = reconcile()
assert checks_of(again)[("numbering_gaps", "INV/2026")]["explanation"] == "deleted test invoice"
must(api("POST", f"/migration/runs/{RUN}/sign", OWNER, {}))
s, b = load()
assert s == 409 and "new run" in b["detail"]
s, b = api("POST", f"/migration/runs/{RUN}/discard", OWNER, {"reason": "x"})
assert s == 409
def test_a_customer_with_unsigned_history_is_not_invoiced_until_the_owner_signs(odoo, world):
"""Invoicing takes writes for these customers from the signature on."""
must(load())
s, b = api("POST", "/invoices", BILLER, {"bp_id": str(BP), "currency": "GBP", "lines": [line()]})
assert s == 409 and "not yet signed off" in b["detail"]
migrate_payments(world, odoo)
reconcile()
explain_and_sign(world)
assert api("POST", "/invoices", BILLER, {"bp_id": str(BP), "currency": "GBP", "lines": [line()]})[0] == 200
def test_discarding_a_run_removes_its_documents(odoo, engine):
must(load())
out = must(api("POST", f"/migration/runs/{RUN}/discard", OWNER, {"reason": "wrong cutover date"}))
assert out["status"] == "discarded" and out["documents_removed"] == 5
assert counts(engine)["invoices"] == 0 and counts(engine)["credits"] == 0
assert api("GET", "/invoices/odoo:odooprod:100", BILLER)[0] == 404
assert api("POST", f"/migration/runs/{RUN}/sign", OWNER, {})[0] == 409
def test_the_load_and_reconcile_are_machine_permissions_and_the_run_is_readable_by_the_owner(odoo):
for who in (BILLER, APPROVER, OWNER):
assert api("POST", "/migration/odoo/load", who, {"cutover_date": "2026-10-31"})[0] == 403
assert api("POST", f"/migration/runs/{RUN}/reconcile", who, {})[0] == 403
must(load())
assert must(api("GET", "/migration/runs", OWNER))["runs"][0]["run_id"] == RUN
assert api("GET", f"/migration/runs/{RUN}", BILLER)[0] == 403
assert person("x@y", ["ar:migration:load"])
def test_an_unconfigured_odoo_is_a_clear_503_not_a_crash(monkeypatch, world):
from brick_ar.application import use_cases
monkeypatch.undo()
for k in ("ODOO_URL", "ODOO_DB", "ODOO_USER", "ODOO_KEY"):
monkeypatch.delenv(k, raising=False)
s, b = load()
assert s == 503 and "ODOO_URL" in b["detail"]
assert use_cases.odoo_reader