dodil/erp-brick-expenses
DODIL ERP brick: expenses
tests/test_migration.py
"""Slice B item 5 (last): the Odoo slice (EXP-24, workflow exp_migration_run): posted and paid sheets as read-only claims, the unpaid ones to be paid here, the
receipt images as objects, reconciled to Odoo and signed by the owner. Odoo is read through a port; here it answers fake rows (no live Odoo)."""
import base64
import pytest
from conftest import EMPLOYEE, FINANCE, JPEG, MANAGER, OWNER, PAYMENTS_CLIENT, SYSTEM, api, must, refused
from brick_expenses.application import use_cases
from brick_expenses.infrastructure import odoo as odoo_port
DB = "acme"
EMPLOYEES = {"11": 3301, "12": 2280}
PRODUCTS = {"TRAVEL": "travel", "HOTEL": "accommodation"}
def sheet(sid, emp, state, total, when="2026-09-30", payment="not_paid", **kw):
return {"id": sid, "name": f"Expenses {sid}", "employee_id": [emp, f"Employee {emp}"], "company_id": [1, "Acme"], "state": state, "payment_state": payment,
"user_id": [2, "Manager"], "total_amount": total, "currency_id": [1, "GBP"], "accounting_date": when if state in ("post", "done") else False,
"approval_date": when, **kw}
def expense(eid, sid, emp, product, total, mode="own_account", name="Item", when="2026-09-20", cur=None, orig=None):
return {"id": eid, "name": name, "sheet_id": [sid, "S"], "employee_id": [emp, "E"], "product_id": [product, "P"], "date": when,
"total_amount_currency": orig if orig is not None else total, "currency_id": [2, cur or "GBP"], "total_amount": total, "tax_ids": [], "payment_mode": mode,
"description": False, "state": "done"}
class FakeOdoo:
def __init__(self):
self.currency = "GBP"
self.down = None
self.sheet_rows = [
sheet(1, 11, "done", 86.40, "2025-06-30", "paid"),
sheet(2, 12, "done", 150.00, "2026-01-15", "paid"),
sheet(3, 11, "post", 200.00, "2026-09-30"), # posted in Odoo, not paid: to be paid here
sheet(4, 11, "approve", 55.55, "2026-10-20"), # approved, never posted in Odoo
sheet(5, 12, "approve", 40.00, "2026-10-21"),
sheet(6, 11, "submit", 10.00, "2026-10-22"), # not carried
sheet(7, 11, "draft", 5.00, "2026-10-23"),
sheet(8, 12, "cancel", 5.00, "2026-10-24"),
]
self.expense_rows = [
expense(101, 1, 11, 501, 86.40, name="Northern Rail"),
expense(102, 2, 12, 501, 120.00, name="Train"), expense(103, 2, 12, 502, 30.00, mode="company_account", name="Hotel breakfast"),
expense(104, 3, 11, 502, 200.00, name="Hotel", cur="EUR", orig=230.00),
expense(105, 4, 11, 501, 55.55), expense(106, 5, 12, 501, 40.00),
expense(107, 6, 11, 501, 10.00), expense(108, 7, 11, 501, 5.00), expense(109, 8, 12, 501, 5.00),
]
self.product_rows = [{"id": 501, "default_code": "TRAVEL", "name": "Travel"}, {"id": 502, "default_code": "HOTEL", "name": "Hotel"}]
self.attachment_rows = [{"id": 901, "res_id": 101, "name": "ticket.jpg", "mimetype": "image/jpeg", "datas": base64.b64encode(JPEG).decode()},
{"id": 902, "res_id": 104, "name": "notes.txt", "mimetype": "text/plain", "datas": base64.b64encode(b"just text").decode()}]
def _check(self):
if self.down:
raise odoo_port.OdooUnavailable(502, self.down)
def company_currency(self, db):
self._check()
return self.currency
def sheets(self, db, years):
self._check()
return [s for s in self.sheet_rows if int(str(s.get("accounting_date") or s["approval_date"])[:4]) in years]
def expenses(self, db, sheet_ids):
return [e for e in self.expense_rows if e["sheet_id"][0] in sheet_ids]
def products(self, db):
return self.product_rows
def attachments(self, db, expense_ids):
return [a for a in self.attachment_rows if a["res_id"] in expense_ids]
def attachment_count(self, db, expense_ids):
return len(self.attachments(db, expense_ids))
@pytest.fixture
def odoo():
fake = FakeOdoo()
odoo_port.ports.reader = fake
yield fake
odoo_port.ports.reader = None
def load_body(**kw):
return {"odoo_db": DB, "cutover_date": "2026-11-07", "company_code": "UK01", "years": [2025, 2026], "employee_map": EMPLOYEES, "product_purposes": PRODUCTS,
"upstream_signed": True, "drafts": "finished_in_odoo", **kw}
def load(who=None, **kw):
return api("POST", "/migration/odoo/load", who or SYSTEM, load_body(**kw))
RUN = f"odoo:{DB}:2026-11-07"
def claim(rid, who=None):
return api("GET", f"/reports/{rid}", who or FINANCE)
def reconcile(who=None, run=RUN):
return api("POST", f"/migration/odoo/{run}/reconcile", who or SYSTEM, {"employee_map": EMPLOYEES})
# ── the load ─────────────────────────────────────────────────────────────────────────────────────
def test_EXP_24_posted_and_paid_sheets_become_read_only_claims_and_the_unpaid_ones_are_paid_here(nb, odoo, store):
got = must(load())
assert got["run_id"] == RUN and got["status"] == "loaded" and got["stats"]["sheets"] == 5 and got["stats"]["claims_written"] == 5 and got["stats"]["lines"] == 6
ids = {i: f"odoo:{DB}:hr.expense.sheet:{i}" for i in range(1, 9)}
states = {i: must(claim(ids[i]))["status"] for i in range(1, 6)}
assert states == {1: "paid", 2: "paid", 3: "posted", 4: "posted", 5: "posted"} # done = paid, post = posted; the approved ones were posted by the normal path
for i in (6, 7, 8):
refused(claim(ids[i]), 404) # submitted, draft and cancelled sheets are not carried
s2 = must(claim(ids[2]))
assert (s2["total_amount"], s2["reimbursable_amount"], s2["currency"], s2["company_code"], s2["person_id"], s2["approved_on"]) == ("150.00", "120.00", "GBP", "UK01", "2280", "2026-01-15")
assert [(ln["paid_with"], ln["amount"], ln["category_code"]) for ln in s2["lines"]] == [("own", "120.00", "TRAVEL"), ("card", "30.00", "HOTEL")]
s3 = must(claim(ids[3]))["lines"][0]
assert (s3["amount"], s3["receipt_amount"], s3["receipt_currency"], s3["vat_amount"]) == ("200.00", "230.00", "EUR", "0.00") # Odoo's own amounts, never re-priced
with use_cases.repository() as repo:
cats = {c["category_code"]: c["gl_purpose"] for c in repo.categories.all_in_force("UK01", use_cases._today(repo))}
assert cats["TRAVEL"] == "travel" and cats["HOTEL"] == "accommodation"
def test_EXP_24_the_gl_and_the_payouts_follow_only_where_odoo_booked_nothing_or_nothing_was_paid(nb, odoo):
must(load())
assert sorted(j["source_document"] for j in nb.journals.values()) == [f"exp:odoo:{DB}:hr.expense.sheet:4:r1", f"exp:odoo:{DB}:hr.expense.sheet:5:r1"] # gl's migration carries 1-3
assert {p["source_document"]: p["amount"] for p in nb.payouts.values()} == {
f"odoo:{DB}:hr.expense.sheet:3": "200.00", f"odoo:{DB}:hr.expense.sheet:4": "55.55", f"odoo:{DB}:hr.expense.sheet:5": "40.00"} # the three unpaid, nothing else
assert all(p["payee_bp_id"] in ("73301", "72280") and p["clearing_purpose"] == "employee_payable" for p in nb.payouts.values())
refs = {p["payout_ref"] for p in nb.payouts.values()}
assert f"exp:odoo:{DB}:hr.expense.sheet:3:r1:pay" in refs
done = must(api("POST", f"/reports/odoo:{DB}:hr.expense.sheet:3/paid", PAYMENTS_CLIENT,
{"payout_ref": f"exp:odoo:{DB}:hr.expense.sheet:3:r1:pay", "payment_id": "BANK-9", "paid_on": "2026-11-12", "amount": "200.00"}))
assert done["status"] == "paid" # paid here, on payments' confirmation
def test_EXP_24_the_claims_are_the_finances_and_the_owners_to_read_and_nobody_can_decide_one(nb, odoo):
must(load())
rid = f"odoo:{DB}:hr.expense.sheet:3"
refused(api("GET", f"/reports/{rid}", EMPLOYEE), 404)
refused(api("GET", f"/reports/{rid}", MANAGER), 404)
assert must(api("GET", f"/reports/{rid}", OWNER))["report_id"] == rid
for action, body in (("approve", None), ("return", {"reason": "x"}), ("reject", {"reason": "x"})):
assert api("POST", f"/reports/{rid}/{action}", FINANCE, body)[0] in (404, 409), action # history is read-only: nothing to decide
refused(api("POST", f"/reports/{rid}/lines", FINANCE, {"category_code": "TRAVEL", "expense_date": "2026-11-01", "merchant": "x", "amount": "1.00"}), 404)
def test_EXP_24_receipt_images_become_objects_under_the_documented_key_and_what_is_not_an_image_is_named(nb, odoo, store):
got = must(load())
assert got["stats"]["images"] == 1 and got["stats"]["images_skipped"] == ["attachment 902: not a receipt image"]
key = next(k for k in store.objects if k.startswith("raw/expenses/"))
assert key.startswith(f"raw/expenses/odoo:{DB}:hr.expense.sheet:1/1/") and key.endswith(".jpg") and store.objects[key][0] == JPEG
line = must(claim(f"odoo:{DB}:hr.expense.sheet:1"))["lines"][0]
assert line["receipt_key"] == key
assert must(api("GET", f"/reports/odoo:{DB}:hr.expense.sheet:1/lines/1/receipt", FINANCE))["size_bytes"] == len(JPEG)
def test_EXP_24_a_second_load_writes_nothing_new(nb, odoo):
first = must(load())
posts = (nb.journal_posts, nb.payout_posts)
again = must(load())
assert (again["stats"]["claims_written"], again["stats"]["claims_unchanged"], again["stats"]["lines"]) == (0, 5, 0) and again["status"] == "loaded"
assert (nb.journal_posts, nb.payout_posts) == posts and len(must(api("GET", "/reports", FINANCE, {"scope": "all"}))["reports"]) == 5 == first["stats"]["claims_written"]
with use_cases.repository() as repo:
assert len(repo.migration.of_run(RUN, "hr.expense.sheet")) == 5
def test_EXP_24_a_sheet_loaded_after_odoo_gained_one_is_added_and_the_run_must_be_reconciled_again(nb, odoo):
must(load())
must(reconcile())
odoo.sheet_rows.append(sheet(9, 12, "done", 12.00, "2026-10-30", "paid"))
odoo.expense_rows.append(expense(110, 9, 12, 501, 12.00))
got = must(load())
assert got["stats"]["claims_written"] == 1 and got["status"] == "loaded" and got["checks"] == []
def test_EXP_24_the_doors_are_answered_for_everything_odoo_holds_or_nothing_is_loaded(nb, odoo):
body = refused(load(employee_map={"11": 3301}), 409, "Odoo employee 12 has no person in employee_map")
assert body["gate"] == "doors_decided"
refused(load(product_purposes={"TRAVEL": "travel"}), 409, "Odoo product 502 (Hotel) has no category")
odoo.product_rows[0]["default_code"] = False
refused(load(), 409, "Odoo product 501 (Travel) has no category")
odoo.product_rows[0]["default_code"] = "TRAVEL"
refused(load(upstream_signed=False), 409, "must be signed first")
refused(load(years=[]), 422)
refused(load(years=[1999]), 422)
refused(load(drafts="whatever"), 422)
assert must(api("GET", "/reports", FINANCE, {"scope": "all"}))["reports"] == [] and nb.journals == {} # nothing half-loaded
def test_EXP_24_no_odoo_connection_or_odoo_down_or_a_sheet_that_does_not_add_up_is_a_reply(nb, odoo):
odoo.down = "Odoo did not answer"
refused(load(), 502, "did not answer")
odoo.down = None
odoo.expense_rows[0]["total_amount"] = 80.00
body = refused(load(), 409, "lines add up to 80.00 but Odoo says 86.40")
assert body["gate"] == "sheet_total"
odoo_port.ports.reader = None
refused(load(), 409, "no Odoo connection is configured")
def test_EXP_24_an_employee_hr_gives_no_partner_is_listed_unpayable_not_silently_dropped(nb, odoo):
nb.persons["emp3301@uk.example"]["bp_id"] = None
got = must(load())
assert any("person 3301 no person partner" in u for u in got["stats"]["unpayable"])
assert f"odoo:{DB}:hr.expense.sheet:3" not in {p["source_document"] for p in nb.payouts.values()}
def test_EXP_24_the_machine_permissions_are_the_systems_and_the_owners_alone(nb, odoo):
refused(load(EMPLOYEE), 403, "exp:migration:load")
refused(load(FINANCE), 403, "exp:migration:load")
refused(load(OWNER), 403, "exp:migration:load")
must(load())
refused(reconcile(FINANCE), 403, "exp:migration:reconcile")
refused(api("POST", f"/migration/odoo/{RUN}/sign", FINANCE), 403, "exp:migration:sign")
refused(api("POST", f"/migration/odoo/{RUN}/discard", FINANCE, {"reason": "x"}), 403, "exp:migration:sign")
refused(api("GET", f"/migration/odoo/{RUN}", EMPLOYEE), 403, "exp:migration:read")
assert must(api("GET", f"/migration/odoo/{RUN}", FINANCE))["status"] == "loaded"
assert [r["status"] for r in must(api("GET", "/migration/odoo", OWNER))["runs"]] == ["loaded"]
# ── reconcile, sign ──────────────────────────────────────────────────────────────────────────────
def test_EXP_24_what_was_loaded_equals_odoo_to_the_penny_per_employee_per_month_and_per_image(nb, odoo):
must(load())
got = must(reconcile())
assert got["status"] == "reconciled"
by = {c["check"]: c for c in got["checks"]}
assert all(c["equal"] for c in got["checks"]) is False # one image is not carried: attachment 902 is text
assert by["unpaid_reimbursable_per_employee"]["equal"] and by["unpaid_reimbursable_per_employee"]["odoo"] == "295.55"
assert by["unpaid_reimbursable_per_employee"]["detail"] == {"11": {"odoo": "255.55", "ours": "255.55"}, "12": {"odoo": "40.00", "ours": "40.00"}}
assert by["claims_and_totals_per_month"]["equal"] and by["claims_and_totals_per_month"]["detail"]["2026-10"] == {"odoo": "2 / 95.55", "ours": "2 / 95.55"}
assert (by["receipt_images_carried"]["odoo"], by["receipt_images_carried"]["ours"], by["receipt_images_carried"]["equal"]) == ("2", "1", False)
assert [g["met"] for g in got["readiness"]] == [False, False]
def test_EXP_24_the_owner_signs_only_when_every_check_is_equal_or_explained_and_a_signed_run_is_frozen(nb, odoo):
must(load())
must(reconcile())
body = refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 409, "receipt_images_carried")
assert body["gate"] == "checks_reconciled" and [g["met"] for g in body["gates"]] == [False, False]
refused(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "receipt_images_carried", "reason": " "}), 422)
refused(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "claims_and_totals_per_month", "reason": "x"}), 409, "nothing to explain")
refused(api("POST", f"/migration/odoo/{RUN}/explain", FINANCE, {"check": "receipt_images_carried", "reason": "x"}), 403, "exp:migration:sign")
got = must(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "receipt_images_carried", "reason": "attachment 902 is a text note, not a receipt"}))
assert "text note" in next(c for c in got["checks"] if c["check"] == "receipt_images_carried")["explained"] and [g["met"] for g in got["readiness"]] == [True, True]
signed = must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))
assert signed["status"] == "signed" and signed["signed_by"] == "owner@uk.example" and signed["signed_on"] == "2026-11-07"
assert must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))["status"] == "signed" # one signature
refused(load(), 409, "is signed")
refused(reconcile(), 409, "is signed")
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "x"}), 409, "only a loaded or reconciled run")
refused(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "receipt_images_carried", "reason": "x"}), 409, "only a reconciled run")
def test_EXP_24_a_difference_odoo_made_after_the_load_fails_the_check_and_is_found_by_the_next_reconcile(nb, odoo):
odoo.attachment_rows.pop() # no stray attachment: images will be equal
must(load())
first = must(reconcile())
assert all(c["equal"] for c in first["checks"]) and [g["met"] for g in first["readiness"]] == [True, True]
odoo.sheet_rows[2]["payment_state"] = "paid" # sheet 3 was paid in Odoo after our load
odoo.sheet_rows[2]["state"] = "done"
again = must(reconcile())
by = {c["check"]: c for c in again["checks"]}
assert by["unpaid_reimbursable_per_employee"]["equal"] is False
assert by["unpaid_reimbursable_per_employee"]["detail"]["11"] == {"odoo": "55.55", "ours": "255.55"}
refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 409, "unpaid_reimbursable_per_employee")
def test_EXP_24_reconcile_needs_a_run_that_exists_and_is_open(nb, odoo):
refused(reconcile(), 404, "no migration run")
refused(api("GET", f"/migration/odoo/{RUN}", FINANCE), 404)
refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 404)
must(load())
refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 409, "only a reconciled run is signed")
# ── discard ──────────────────────────────────────────────────────────────────────────────────────
def test_EXP_24_a_run_nothing_has_left_for_is_discarded_with_a_reason_and_taken_back_out(nb, odoo):
nb.gl_status, nb.pay_status = 500, 500 # the neighbours are down: every row stays pending
must(load())
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": " "}), 422)
got = must(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "wrong employee map"}))
assert got["status"] == "discarded" and got["discard_reason"] == "wrong employee map"
assert must(api("GET", "/reports", FINANCE, {"scope": "all"}))["reports"] == []
with use_cases.repository() as repo:
assert repo.migration.of_run(RUN) == [] and repo.outbox.pending() == []
nb.gl_status = nb.pay_status = None
again = must(load())
assert again["run_id"] == f"{RUN}:2" and again["stats"]["claims_written"] == 5 # a discarded run is terminal: the next is a new run
assert [r["status"] for r in must(api("GET", "/migration/odoo", OWNER))["runs"]] == ["discarded", "loaded"]
def test_EXP_24_a_run_whose_journals_or_payouts_have_left_cannot_be_discarded(nb, odoo):
must(load())
body = refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "oops"}), 409, "already left for gl or payments")
assert body["gate"] == "nothing_sent"
assert must(api("GET", f"/migration/odoo/{RUN}", OWNER))["status"] == "loaded"
# ── the map ─────────────────────────────────────────────────────────────────────────────────────
def test_EXP_24_the_one_map_route_answers_what_was_migrated_and_omits_what_was_not(nb, odoo):
must(load())
got = must(api("GET", f"/migration-map/{DB}/hr.expense.sheet", FINANCE, {"ids": "1,3,99"}))
assert got == [{"odoo_model": "hr.expense.sheet", "odoo_id": "1", "ours": f"odoo:{DB}:hr.expense.sheet:1"},
{"odoo_model": "hr.expense.sheet", "odoo_id": "3", "ours": f"odoo:{DB}:hr.expense.sheet:3"}]
lines = must(api("GET", f"/migration-map/{DB}/hr.expense", FINANCE, {"ids": "102,103"}))
assert [x["ours"] for x in lines] == [f"odoo:{DB}:hr.expense.sheet:2:1:1", f"odoo:{DB}:hr.expense.sheet:2:1:2"]
assert must(api("GET", f"/migration-map/{DB}/ir.attachment", FINANCE, {"ids": "901"}))[0]["ours"].startswith("raw/expenses/")
assert must(api("GET", "/migration-map/other/hr.expense.sheet", FINANCE, {"ids": "1"})) == []
refused(api("GET", f"/migration-map/{DB}/hr.expense.sheet", FINANCE, {"ids": "1,x"}), 422, "comma-separated digits")
refused(api("GET", f"/migration-map/{DB}/hr.expense.sheet", FINANCE, {"ids": ""}), 422)
refused(api("GET", f"/migration-map/{DB}/res.partner", FINANCE, {"ids": "1"}), 422)
refused(api("GET", f"/migration-map/{DB}/hr.expense.sheet", EMPLOYEE, {"ids": "1"}), 403, "exp:migration:read")