dodil/erp-brick-payments
DODIL ERP brick: payments
tests/test_migration.py
"""The Odoo migration (PAY-44, PAY-15, SC-PAY-09; module.yaml migration.odoo): a reader port with fake Odoo rows, the id
map, an idempotent re-run, the reconcile checks as {check, odoo, ours, equal, explanation}, history read-only once signed."""
import json
from decimal import Decimal
import pytest
from conftest import APPROVER, CASHIER, SYSTEM, api, must, person, receipt, refused, settle
from odoo_fake import CUTOVER, DB, SUSPENSE, FakeOdoo, FakePartners, load_ar
from brick_payments.application import use_cases
from brick_payments.infrastructure import odoo
@pytest.fixture(autouse=True)
def fakes(nb):
reader = FakeOdoo()
load_ar(nb)
odoo.use(odoo.Ports(reader, FakePartners()))
yield reader
odoo.use(None)
def run(**kw):
return api("POST", "/migration/odoo/run", SYSTEM, {"db": DB, "cutover_date": CUTOVER, "suspense_bp_id": SUSPENSE, **kw})
def a(model, odoo_id):
return f"odoo:{DB}:{model}:{odoo_id}"
def table_counts():
from sqlalchemy import func, select
from brick_payments.infrastructure import tables as t
with use_cases.repository() as repo:
return {tb.name: repo.conn.execute(select(func.count()).select_from(tb)).scalar() for tb in t.md.sorted_tables}
def test_PAY_44_the_run_loads_everything_odoo_holds_with_odoos_own_dates(nb):
got = must(run())
assert (got["bank_accounts"], got["receipts"], got["unreconciled_credits"], got["refunds"], got["allocations"], got["credit_limits"],
got["followup_levels"]) == (1, 7, 3, 1, 6, 1, 1)
assert got["problems"] == [] and got["skipped_detail"] == []
rec = must(api("GET", f"/receipts/{a('account.payment', 1)}", CASHIER))
assert (rec["received_on"], rec["amount"], rec["bp_id"], rec["origin"], rec["bank_account_id"]) == (
"2026-10-10", "600.00", "7311", a("account.payment", 1), "BNK1") # Odoo's date: the one place rows are dated in the past
assert rec["unallocated"] == "0.00" and rec["allocations"][0]["allocated_on"] == "2026-10-10" and rec["allocations"][0]["origin"] == a("account.partial.reconcile", 1)
s = settle("odoo:prod1:101", "odoo:prod1:102", "odoo:prod1:104", "odoo:prod1:105")
assert {k: v["outstanding"] for k, v in s.items()} == {"odoo:prod1:101": "300.00", "odoo:prod1:102": "0.00", "odoo:prod1:104": "0.00",
"odoo:prod1:105": "0.00"}
assert s["odoo:prod1:101"]["state"] == "part_paid" and s["odoo:prod1:102"]["state"] == "paid"
def test_PAY_44_unreconciled_bank_credits_are_unallocated_receipts_on_the_named_customer_else_the_suspense_customer(nb):
must(run())
named = must(api("GET", f"/receipts/{a('account.bank.statement.line', 1)}", CASHIER))
assert (named["bp_id"], named["unallocated"], named["allocation_state"]) == ("7312", "80.00", "unallocated")
assert must(api("GET", f"/receipts/{a('account.bank.statement.line', 3)}", CASHIER))["bp_id"] == str(SUSPENSE)
items = must(api("GET", "/customers/7312/open-items", CASHIER))
assert [c["amount"] for c in items["credits"]] == ["-80.00"]
# the cashier clears the suspense money onto the customer who paid
moved = must(api("POST", f"/receipts/{a('account.bank.statement.line', 3)}/assign-customer", CASHIER, {"bp_id": 7311}))
assert moved["bp_id"] == "7311"
def test_PAY_44_a_cross_currency_payment_is_carried_in_the_invoices_currency_with_its_rate(nb):
must(run())
r = must(api("GET", f"/receipts/{a('account.payment', 3)}", CASHIER))
assert (r["currency"], r["amount"], r["fx_rate"], r["unallocated"]) == ("EUR", "1000.00", "1.1765", "0.00")
alloc = r["allocations"][0]
assert (alloc["amount"], alloc["currency"]) == ("850.00", "GBP") and settle("odoo:prod1:104")["odoo:prod1:104"]["outstanding"] == "0.00"
def test_PAY_44_a_refund_credit_limit_and_followup_level_are_carried(nb):
must(run())
ref = must(api("GET", f"/refunds/{a('account.payment', 5)}", CASHIER))
assert (ref["status"], ref["amount"], ref["paid_on"], ref["requested_by"], ref["approved_by"]) == ("paid", "50.00", "2026-10-25", "odoo", "odoo")
p4 = must(api("GET", f"/receipts/{a('account.payment', 4)}", CASHIER))
assert p4["refunded"] == "50.00" and p4["unallocated"] == "0.00"
limit = must(api("GET", "/customers/7311/credit", APPROVER))["limits"]
assert [(x["valid_from"], x["amount"], x["origin"]) for x in limit] == [(CUTOVER, "5000.00", a("res.partner", 11))]
od = must(api("GET", "/collections/overdue", APPROVER and CASHIER))["invoices"]
assert [(i["invoice_id"], i["level_reached"]) for i in od] == [("odoo:prod1:101", 2)] # reminders do not restart at level 1
must(api("POST", "/reminder-levels", APPROVER, {"level": {"level": 1, "days_after_due": 7, "mode": "suggest"}}))
must(api("POST", "/reminder-levels", APPROVER, {"level": {"level": 2, "days_after_due": 14, "mode": "suggest"}}))
assert must(api("POST", "/collections/tick", SYSTEM, {}))["drafted"] == 0 # level 2 already reached
settings = {s["setting_key"]: s["value"] for s in must(api("GET", "/settings", CASHIER))["settings"]}
assert settings["statement_import_from"] == "2026-11-16" and settings["suspense_bp_id"] == SUSPENSE
def test_PAY_44_the_second_run_adds_nothing(nb):
must(run())
before = table_counts()
again = must(run())
assert (again["bank_accounts"], again["receipts"], again["refunds"], again["allocations"], again["credit_limits"], again["followup_levels"]) == (0,) * 6
assert again["already_loaded"] >= 15
after = table_counts()
assert {k: (before[k], after[k]) for k in before if before[k] != after[k]} == {"pay_migration_run": (1, 2)} # only the re-run's own record (audit 1)
def test_a_statement_overlapping_the_cutover_skips_the_migrated_lines(nb):
must(run())
csv = "booking_date,amount,currency,entry_ref\n2026-11-10,25.00,GBP,UIID-2\n2026-11-17,10.00,GBP,NEW-1\n"
from conftest import bank
bank("BNK1")
got = must(api("POST", "/receipts/import", CASHIER, {"format": "csv", "content": csv, "bank_account_id": "BNK1"}))
assert (got["created"], got["skipped_before_cutover"]) == (1, 1)
def test_SC_PAY_09_the_reconciliation_is_equal_everywhere_but_the_one_explained_difference(nb):
must(run())
rec = must(api("GET", "/migration/reconcile", APPROVER))
checks = {r["check"] for r in rec["checks"]}
assert checks == {"open balance per customer and currency", "the same open balance as ar's migration", "outstanding per invoice",
"unallocated per receipt", "receipts per bank account and month", "allocated total", "paid invoices", "refunds paid",
"cross-currency payments"}
unequal = [r for r in rec["checks"] if not r["equal"]]
assert [(r["check"], r["subject"]) for r in unequal] == [("cross-currency payments", a("account.payment", 3))]
x = unequal[0]
assert (x["odoo"], x["ours"], x["must"]) == ("1000.00 EUR", "850.00 GBP", "explained") and "1.1765" in x["explanation"] and "ledger" in x["explanation"]
assert rec["blocking"] == 0 and rec["can_sign"] is True and rec["explained_differences"] == 1
by = {(r["check"], r["subject"]): r for r in rec["checks"]}
assert (by["open balance per customer and currency", "7311 GBP"]["odoo"], by["open balance per customer and currency", "7311 GBP"]["ours"]) == ("275.00", "275.00")
assert by["open balance per customer and currency", "7312 GBP"]["ours"] == "-80.00" # money received and not yet matched
assert by["receipts per bank account and month", "BNK1 2026-10"]["odoo"] == [4, "2350.00"]
assert all(set(r) >= {"check", "odoo", "ours", "equal", "explanation"} for r in rec["checks"])
assert must(api("GET", "/migration/reconcile", APPROVER, {"only_differences": True}))["checks"] == unequal
def test_PAY_15_a_check_that_must_equal_and_does_not_blocks_the_signature(nb, fakes):
must(run())
nb.invoices["odoo:prod1:101"]["total"] = "1100.00" # ar's side drifted after the run
rec = must(api("GET", "/migration/reconcile", APPROVER))
assert rec["can_sign"] is False and rec["blocking"] >= 3
bad = {(r["check"], r["subject"]) for r in rec["checks"] if r["must"] == "equal" and not r["equal"]}
assert ("outstanding per invoice", "odoo:prod1:101") in bad and ("open balance per customer and currency", "7311 GBP") in bad
body = refused(api("POST", "/migration/sign", APPROVER, {}), 409, "not equal and not explained")
assert "odoo:prod1:101" in json.dumps(body)
nb.invoices["odoo:prod1:101"]["total"] = "1000.00"
assert must(api("GET", "/migration/reconcile", APPROVER))["can_sign"] is True
def test_PAY_15_the_owner_signs_and_the_signature_records_the_figures_and_freezes_the_migration(nb):
must(run())
assert api("POST", "/migration/sign", CASHIER, {})[0] == 403
signed = must(api("POST", "/migration/sign", APPROVER, {}))
assert signed["signed"]["by"] == "approver@shop.example"
again = must(api("POST", "/migration/sign", APPROVER, {}))
assert again["unchanged"] is True and again["signed"]["by"] == "approver@shop.example"
with use_cases.repository() as repo:
run_row = repo.migration.run()
assert run_row["signed_by"] == "approver@shop.example" and len(run_row["signed_figures"]) == len(signed["checks"])
refused(run(), 409, "is read-only now")
assert must(api("GET", "/migration/reconcile", APPROVER))["signed"]["by"] == "approver@shop.example" # still readable
def test_the_gates_partners_mapped_invoices_loaded_and_suspense_customer(nb, fakes):
odoo.use(odoo.Ports(fakes, FakePartners({11: 7311})))
refused(run(), 409, "not in business-partner's migration map: [12]")
odoo.use(odoo.Ports(fakes, FakePartners()))
refused(run(suspense_bp_id=None), 409, "name no customer")
del nb.invoices["odoo:prod1:104"]
body = refused(run(), 409, "settle documents ar does not have")
assert "odoo:prod1:104" in json.dumps(body)
assert table_counts()["pay_receipt"] == 0 and table_counts()["pay_allocation"] == 0 # nothing was written
odoo.use(odoo.Ports(None, None))
refused(run(), 409, "No Odoo database is configured")
def test_a_receipt_odoo_dated_in_the_future_is_reported_not_loaded_and_blocks_the_signature(nb, fakes):
fakes.rows["account.payment"][0]["date"] = "2026-12-25"
got = must(run())
assert any("after today" in p["why"] for p in got["problems"]) and got["receipts"] == 6
assert must(api("GET", "/migration/reconcile", APPROVER))["can_sign"] is False
def test_the_migration_is_a_machine_and_never_a_persons_wildcard(nb):
body = {"db": DB, "cutover_date": CUTOVER}
for who in (person("star@shop.example", ["pay:*:*"]), person("all@shop.example", ["*:*:*"]), APPROVER, CASHIER):
assert api("POST", "/migration/odoo/run", who, body)[0] == 403
assert api("POST", "/migration/odoo/run", SYSTEM, {**body, "cutover_date": "2026-11-20"})[0] == 409 # a day that has not ended
assert api("GET", "/migration/reconcile", CASHIER)[0] == 403 and api("POST", "/migration/sign", SYSTEM, {})[0] == 403
def test_the_migration_writes_no_ar_table_and_receipts_made_afterwards_are_dated_today_again(nb):
must(run())
nb_invoices = dict(nb.invoices)
r = receipt(bp_id=7311, amount="10.00", reference="after", received_on="2026-11-20")
assert nb.invoices == nb_invoices # payments only reads ar
assert Decimal(r["amount"]) == Decimal("10.00")
refused(api("POST", "/receipts", CASHIER, {"bp_id": 7311, "amount": "1.00", "currency": "GBP", "received_on": "2026-11-21"}), 409, "after today")
def test_later_bricks_read_the_odoo_map_through_one_route(nb):
"""shared.yaml conventions.odoo_migration: GET /migration-map/{db}/{model}?ids= -> [{odoo_model, odoo_id, ours}]; an
unmapped id is left out. The migration's machine permission reads it."""
must(run())
got = must(api("GET", f"/migration-map/{DB}/account.payment", SYSTEM, {"ids": "1,999"}))
assert got == [{"odoo_model": "account.payment", "odoo_id": "1", "ours": a("account.payment", 1)}]