dodil/erp-brick-tax
DODIL ERP brick: tax
tests/test_migration.py
"""The Odoo migration slice (TAX-15, SC-TAX-08): fake Odoo rows through the reader port — no live Odoo. The second run writes
nothing; history is read-only; the reconcile checks are {check, odoo, ours, equal, explanation}."""
import pytest
from conftest import ADMIN, BILLER, SYSTEM, api, compute, line, must, person, tool
from brick_tax.infrastructure import odoo
DB = "acme_prod"
CUTOVER = "2026-11-20"
class FakeOdoo:
"""The Odoo company the spec's given describes: sale taxes 20% S, 0% Z, Exempt, a reverse-charge tax, an archived 5% one
posted invoices used and an archived one nothing used; a purchase tax; the fiscal positions UK, EU B2B, Rest of world."""
def __init__(self):
tag = lambda i, n: {"id": i, "name": n, "applicability": "taxes"} # noqa: E731
self.tables = {
"res.company": [{"id": 1, "vat": "GB123456789", "country_id": 233, "tax_calculation_rounding_method": "round_per_line"}],
"account.tax": [
{"id": 1, "name": "20% S", "description": "20% VAT", "invoice_label": "VAT 20%", "amount": 20, "amount_type": "percent",
"type_tax_use": "sale", "tax_scope": "service", "active": True, "company_id": 1, "xmlid": "l10n_uk.1_ST11"},
{"id": 2, "name": "0% Z", "description": "Zero rated", "amount": 0, "amount_type": "percent", "type_tax_use": "sale",
"active": True, "company_id": 1, "xmlid": "l10n_uk.1_ST2"},
{"id": 3, "name": "Exempt", "description": "Exempt", "amount": 0, "amount_type": "percent", "type_tax_use": "sale",
"active": True, "company_id": 1},
{"id": 4, "name": "Reverse charge", "description": "20% reverse charge", "invoice_label": "Reverse charge: customer accounts for the VAT", "amount": 20, "amount_type": "percent",
"type_tax_use": "sale", "active": True, "company_id": 1},
{"id": 5, "name": "5% R (archived)", "description": "5% VAT", "amount": 5, "amount_type": "percent", "type_tax_use": "sale",
"active": False, "company_id": 1},
{"id": 6, "name": "15% old (archived)", "amount": 15, "amount_type": "percent", "type_tax_use": "sale", "active": False,
"company_id": 1},
{"id": 7, "name": "20% purchase", "amount": 20, "amount_type": "percent", "type_tax_use": "purchase", "active": True,
"company_id": 1}],
"account.tax.repartition.line": [
{"tax_id": 1, "document_type": "invoice", "repartition_type": "base", "factor_percent": 100, "tag_ids": [60]},
{"tax_id": 1, "document_type": "invoice", "repartition_type": "tax", "factor_percent": 100, "tag_ids": [10]},
{"tax_id": 2, "document_type": "invoice", "repartition_type": "base", "factor_percent": 100, "tag_ids": [60]},
{"tax_id": 3, "document_type": "invoice", "repartition_type": "base", "factor_percent": 100, "tag_ids": [60]},
{"tax_id": 4, "document_type": "invoice", "repartition_type": "base", "factor_percent": 100, "tag_ids": [60]},
{"tax_id": 4, "document_type": "invoice", "repartition_type": "tax", "factor_percent": 100, "tag_ids": [10]},
{"tax_id": 4, "document_type": "invoice", "repartition_type": "tax", "factor_percent": -100, "tag_ids": [40]},
{"tax_id": 5, "document_type": "invoice", "repartition_type": "base", "factor_percent": 100, "tag_ids": [60]},
{"tax_id": 5, "document_type": "invoice", "repartition_type": "tax", "factor_percent": 100, "tag_ids": [10]}],
"account.account.tag": [tag(10, "+1"), tag(40, "-4"), tag(60, "+6")],
"account.fiscal.position": [
{"id": 10, "name": "UK", "vat_required": False, "active": True, "company_id": 1},
{"id": 11, "name": "EU B2B", "vat_required": True, "active": True, "company_id": 1},
{"id": 12, "name": "Rest of world", "vat_required": False, "active": True, "company_id": 1}],
"account.fiscal.position.tax": [
{"id": 21, "position_id": 11, "tax_src_id": 1, "tax_dest_id": 2},
{"id": 22, "position_id": 12, "tax_src_id": 1, "tax_dest_id": 2}],
"product.template": [{"id": 100, "taxes_id": [1], "sale_ok": True}, {"id": 101, "taxes_id": [2], "sale_ok": True},
{"id": 102, "taxes_id": [3], "sale_ok": True}, {"id": 103, "taxes_id": [4], "sale_ok": True}],
"res.partner": [
{"id": 1, "vat": "GB111111111", "is_company": True, "property_account_position_id": None, "customer_rank": 1},
{"id": 2, "vat": None, "is_company": False, "property_account_position_id": None, "customer_rank": 1},
{"id": 3, "vat": "IE6388047V", "is_company": True, "property_account_position_id": 11, "customer_rank": 1},
{"id": 4, "vat": "FR12345678901", "is_company": True, "property_account_position_id": 12, "customer_rank": 1}],
"account.move.line": [
{"id": 1, "move_id": 900, "date": "2026-10-01", "move_type": "out_invoice", "state": "posted", "tax_ids": [1], "tax_line_id": None,
"tax_tag_ids": [60], "balance": -100, "tax_base_amount": 0, "product_tmpl_id": 100, "partner_id": 1},
{"id": 2, "move_id": 900, "date": "2026-10-01", "move_type": "out_invoice", "state": "posted", "tax_ids": [], "tax_line_id": 1,
"tax_tag_ids": [10], "balance": -20, "tax_base_amount": 100, "product_tmpl_id": None, "partner_id": 1},
{"id": 3, "move_id": 901, "date": "2026-10-02", "move_type": "out_invoice", "state": "posted", "tax_ids": [2], "tax_line_id": None,
"tax_tag_ids": [60], "balance": -50, "tax_base_amount": 0, "product_tmpl_id": 100, "partner_id": 1}, # hand-edited: 0% on a 20% product
{"id": 4, "move_id": 902, "date": "2026-10-03", "move_type": "out_invoice", "state": "posted", "tax_ids": [], "tax_line_id": 5,
"tax_tag_ids": [10], "balance": -5, "tax_base_amount": 100, "product_tmpl_id": None, "partner_id": 1},
{"id": 5, "move_id": 903, "date": "2024-01-01", "move_type": "out_invoice", "state": "posted", "tax_ids": [1], "tax_line_id": None,
"tax_tag_ids": [], "balance": -1, "tax_base_amount": 0, "product_tmpl_id": 100, "partner_id": 1}], # outside the window
}
self.calls = []
def read(self, model, fields, where=None, include_archived=False):
self.calls.append(model)
rows = self.tables.get(model, [])
return [dict(r) for r in rows if include_archived or r.get("active", True)]
@pytest.fixture
def odoo_rows(monkeypatch):
fake = FakeOdoo()
monkeypatch.setattr(odoo, "reader", fake)
return fake
DOOR = {"odoo_db": DB, "company_id": 1, "registration_from": "2011-01-04", "cutover_date": CUTOVER}
def migrate(who=SYSTEM, **kw):
return api("POST", "/migration/odoo", who, {"door": {**DOOR, **kw}})
def test_tax_15_without_confirm_it_plans_and_writes_nothing(odoo_rows):
out = must(migrate())
assert out["confirmed"] is False and out["written"] is None
plan = out["plan"]
assert {c["tax_code"]: c["lifecycle"] for c in plan["codes"]} == {"S": "active", "Z": "active", "E": "active", "RC": "active", "R": "retired"}
assert plan["proposed_code_map"]["1"] == "S" and plan["proposed_code_map"]["2"] == "Z"
assert [s["odoo_id"] for s in plan["skipped_taxes"]] == ["6"] # archived and nothing used it
assert must(api("GET", "/codes", ADMIN))["codes"] == [] # nothing was written
assert api("GET", "/registration", ADMIN)[1]["recorded"] is False
def test_sc_tax_08_the_migration_reconciles_and_the_second_run_writes_nothing(odoo_rows):
first = must(migrate(confirm=True))
made = first["written"]["created"]
assert made["codes"] == 5 and made["rates"] == 5 and made["registration"] == 1 and made["settings"] == 1 and made["parties"] == 4
assert made["rules"] >= 8 and made["treatments"] >= 3 and made["categories"] >= 4
codes = {c["tax_code"]: c for c in must(api("GET", "/codes", ADMIN))["codes"]}
assert codes["RC"]["rate_pct"] == "20" and codes["RC"]["charged_pct"] == "0" and codes["RC"]["reporting_box"] == "1,4,6"
assert codes["S"]["reporting_box"] == "1,6" and codes["Z"]["reporting_box"] == "6" and codes["R"]["lifecycle"] == "retired"
assert codes["S"]["lifecycle"] == "active" and "Reverse charge" in must(api("GET", "/codes/RC", ADMIN))["document_wording"]
reg = must(api("GET", "/registration", ADMIN))
assert (reg["status_today"], reg["vat_number"], reg["registrations"][0]["effective_from"]) == ("registered", "GB123456789", "2011-01-04")
checks = {c["check"]: c for c in first["reconcile"]}
assert all(set(c) == {"check", "odoo", "ours", "equal", "explanation"} for c in checks.values())
for name in ("active sale taxes", "rate per code", "reporting box per code", "customers with a VAT number", "treatment per customer"):
assert checks[name]["equal"] is True, checks[name]
parity = checks["determination parity"]
assert parity["equal"] is False and "move 901 line 3" in parity["explanation"] and "move 900" not in parity["explanation"] # explained
assert checks["VAT by code, last 12 months"]["equal"] is False and "ar has not stored" in checks["VAT by code, last 12 months"]["explanation"]
assert all(c["explanation"] for c in first["reconcile"])
again = must(migrate(confirm=True))
assert all(v == 0 for v in again["written"]["created"].values()), again["written"] # the second run writes nothing
assert again["written"]["differs"] == [] and again["reconcile"] == first["reconcile"]
def test_the_first_invoice_after_cutover_is_taxed_as_odoo_would_have(odoo_rows):
must(migrate(confirm=True))
sys_ = lambda lines, bp, **kw: must(api("POST", "/documents/compute", SYSTEM, {"lines": lines, "tax_point": CUTOVER, "bp_id": bp, **kw})) # noqa: E731
uk = sys_([line(1, "100.00")], 101) # Odoo partner 1: a UK business, 20% S
assert (uk["lines"][0]["tax_code"], uk["totals"]["vat"]) == ("S", "20.00")
consumer = sys_([line(1, "100.00", category="zero")], 102) # Odoo partner 2: a consumer, 0% Z
assert consumer["lines"][0]["tax_code"] == "Z"
eu = sys_([line(1, "100.00", supply="services")], 103) # EU B2B position: 20% -> 0% Z
assert (eu["treatment_code"], eu["lines"][0]["tax_code"], eu["totals"]["vat"]) == ("overseas_business", "Z", "0.00")
contractor = sys_([line(1, "1000.00", category="construction_service")], 101)
assert (contractor["lines"][0]["tax_code"], contractor["totals"]["vat_due"], contractor["totals"]["vat"]) == ("RC", "200.00", "0.00")
assert must(api("GET", "/settings", ADMIN))["settings"][0]["value"] == "per_line_nearest" # from cutover
def test_history_is_read_only_an_issued_documents_tax_is_never_recomputed(odoo_rows):
must(migrate(confirm=True))
contents = {"invoice_number": "INV-1", "issue_date": CUTOVER, "supplier_name": "Dodil", "supplier_address": "Leeds",
"customer_name": "Acme", "customer_address": "1 High Street"}
must(compute([line(1, "100.00")], document_ref="INV-1", bp_id=101))
frozen = must(api("POST", "/documents/INV-1/freeze", SYSTEM, {"contents": contents}))
odoo_rows.tables["account.tax"][0]["amount"] = 25 # Odoo changes; the rerun finds a difference ...
again = must(migrate(confirm=True))
assert all(v == 0 for v in again["written"]["created"].values()) # ... and writes nothing
assert must(api("GET", "/documents/INV-1", BILLER)) == frozen # the issued document is untouched
assert not hasattr(odoo_rows, "write") and not hasattr(odoo.OdooReader, "write") # the reader port has no write
def test_a_run_over_existing_rows_reports_a_difference_and_leaves_the_row(odoo_rows):
must(api("POST", "/codes", ADMIN, {"tax_code": "S", "treatment": "standard", "rate_pct": "19", "valid_from": "2011-01-04"}))
must(api("POST", "/codes/S/activate-code", ADMIN, {}))
out = must(migrate(confirm=True))
assert any("tax code S exists with another treatment or rate" in d for d in out["written"]["differs"])
assert must(api("GET", "/codes/S", ADMIN))["rate_pct"] == "19"
def test_the_door_questions_are_asked_not_guessed(odoo_rows):
odoo_rows.tables["account.fiscal.position"].append({"id": 13, "name": "Special deal", "vat_required": False, "active": True, "company_id": 1})
status, body = migrate(confirm=True)
assert status == 409 and "Special deal (id 13)" in body["detail"] and "treatment_map" in body["detail"]
assert must(api("GET", "/codes", ADMIN))["codes"] == [] # the plan failed before anything was written
assert migrate(confirm=True, treatment_map={"13": "uk_business"})[0] == 200
def test_a_customer_needing_a_vat_number_without_one_stops_unless_moved_to_a_consumer_treatment(odoo_rows):
odoo_rows.tables["res.partner"][2]["vat"] = None
status, body = migrate(confirm=True)
assert status == 409 and "need" in body["detail"] and "missing_vat_to" in body["detail"]
out = must(migrate(confirm=True, missing_vat_to="uk_consumer"))
assert out["written"]["created"]["parties"] == 4
def test_a_round_globally_company_must_pick_an_hmrc_method_and_unsupported_taxes_stop(odoo_rows):
odoo_rows.tables["res.company"][0]["tax_calculation_rounding_method"] = "round_globally"
status, body = migrate(confirm=True)
assert status == 409 and "round_globally" in body["detail"]
assert must(migrate(confirm=True, rounding_method="per_invoice_round_down"))["plan"]["rounding"]["rounding_method"] == "per_invoice_round_down"
odoo_rows.tables["account.tax"].append({"id": 8, "name": "Fixed levy", "amount": 2, "amount_type": "fixed", "type_tax_use": "sale",
"active": True, "company_id": 1})
status, body = migrate(confirm=True, rounding_method="per_line_nearest")
assert status == 409 and "Fixed levy" in body["detail"]
assert migrate(confirm=True, rounding_method="per_line_nearest", code_map={"8": "S"})[0] == 200 # replaced by a percent code at the door
def test_a_company_without_a_vat_number_migrates_as_not_registered(odoo_rows):
odoo_rows.tables["res.company"][0]["vat"] = None
must(migrate(confirm=True))
reg = must(api("GET", "/registration", ADMIN))
assert reg["registrations"][0]["status"] == "not_registered" and reg["status_today"] == "not_registered"
def test_the_migration_is_a_machine_permission_and_without_a_reader_it_says_so(monkeypatch):
assert migrate(ADMIN)[0] == 403 and migrate(person("w@x", ["*:*:*"]))[0] == 403
monkeypatch.setattr(odoo, "reader", odoo.UnconfiguredReader())
status, body = migrate()
assert status == 503 and "ODOO_URL" in body["detail"]
def test_reconcile_can_be_asked_again_read_only(odoo_rows):
must(migrate(confirm=True))
out = must(api("POST", "/migration/odoo/reconcile", SYSTEM, {"door": DOOR}))
assert len(out["reconcile"]) == 8 and out["reconcile"][0]["equal"] is True
assert tool(SYSTEM, "reconcile_tax_migration", {"door": DOOR})[0] == "ok"
def test_the_map_keeps_what_odoo_id_became_which_natural_key(odoo_rows, engine):
must(migrate(confirm=True))
from sqlalchemy import select
from brick_tax.infrastructure.tables import tax_migration_map
with engine.begin() as conn:
rows = {(r.odoo_model, r.odoo_id): r.our_key for r in conn.execute(select(tax_migration_map))}
assert rows[("account.tax", "1")] == "S" and rows[("account.tax", "5")] == "R" and ("account.tax", "6") not in rows
assert rows[("account.fiscal.position", "11")] == "overseas_business" and rows[("res.partner", "3")].startswith("103|")
def test_other_bricks_read_the_map_through_one_route(odoo_rows):
"""shared.yaml conventions.odoo_migration: GET /migration-map/{db}/{model}?ids= -> [{odoo_model, odoo_id, ours}]; an
unmapped id is left out."""
must(migrate(confirm=True))
got = must(api("GET", f"/migration-map/{DB}/account.tax", BILLER, {"ids": "1,5,6"}))
assert got == [{"odoo_model": "account.tax", "odoo_id": "1", "ours": "S"}, {"odoo_model": "account.tax", "odoo_id": "5", "ours": "R"}]