dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
tests/test_stories_migration.py
"""PUR-33 / PUR-34: the Odoo migration slice - keyed odoo:<db>:<model>:<id>, idempotent, reconciled, signed and then frozen."""
from conftest import ALICE, BUYER, SUPPLIER_A, api, must, world
def odoo(neighbours, qty_received="10", residual="116.00"):
neighbours.maps.partners[501], neighbours.maps.products[11], neighbours.maps.units[1] = SUPPLIER_A, "COLA-075", "PK24"
neighbours.odoo.rows.update({
"purchase.order": [{"id": 7, "name": "P00007", "state": "purchase", "partner_id": 501, "partner_ref": "SO-1", "date_order": "2026-03-04 10:00:00", "date_approve": "2026-03-04 10:05:00",
"currency_id": "GBP", "amount_untaxed": "180.00", "amount_tax": "36.00", "amount_total": "216.00"}],
"purchase.order.line": [{"id": 70, "order_id": 7, "sequence": 1, "display_type": False, "product_id": 11, "name": "Cola", "product_qty": "10", "product_uom": 1, "price_unit": "18.00",
"price_subtotal": "180.00", "qty_received": qty_received, "qty_invoiced": "10"}],
"stock.picking": [{"id": 90, "name": "WH/IN/1", "state": "done", "purchase_id": 7, "date_done": "2026-03-10 09:00:00", "picking_type_code": "incoming"}],
"stock.move": [{"id": 900, "picking_id": 90, "purchase_line_id": 70, "state": "done", "quantity": "10", "product_uom_qty": "10"}],
"account.move": [{"id": 300, "name": "BILL/2026/0001", "move_type": "in_invoice", "state": "posted", "payment_state": "partial", "partner_id": 501, "commercial_partner_id": 501,
"ref": "INV-5521", "invoice_date": "2026-03-12", "invoice_date_due": "2026-04-11", "date": "2026-03-12", "currency_id": "GBP", "amount_untaxed": "180.00",
"amount_tax": "36.00", "amount_total": "216.00", "amount_residual": residual}],
"account.move.line": [{"id": 3000, "move_id": 300, "display_type": "product", "product_id": 11, "name": "Cola", "quantity": "10", "product_uom_id": 1, "price_unit": "18.00",
"price_subtotal": "180.00", "price_total": "216.00", "purchase_line_id": 70}],
"account.payment": [{"id": 50, "name": "PAY/1", "payment_type": "outbound", "partner_type": "supplier", "partner_id": 501, "amount": "100.00", "currency_id": "GBP",
"date": "2026-03-20", "state": "posted"}]})
def test_pur33_the_load_carries_history_keyed_by_odoo_ids_and_a_second_load_creates_nothing(neighbours):
world()
odoo(neighbours)
r = must(api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"}))
assert r["created"] == {"orders": 1, "receipts": 1, "bills": 1, "payments": 1} and r["run_id"] == "odoo:acme:2026-06-30"
o = must(api("GET", "/purchase-orders/odoo:acme:purchase.order:7", ALICE))
assert o["order_no"] == "P00007" and o["status"] == "confirmed" and str(o["net_total"]) == "180.00"
b = must(api("GET", "/bills/odoo:acme:account.move:300", ALICE))
assert b["status"] == "approved" and b["supplier_invoice_no"] == "INV-5521" and str(b["outstanding"]) == "116.00"
assert neighbours.inventory.moves == {} and neighbours.gl.journals == {} or not neighbours.gl.journals # not handed to inventory, not posted
again = must(api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"}))
assert again["created"] == {"orders": 0, "receipts": 0, "bills": 0, "payments": 0} and again["replay"] is True
assert must(api("GET", "/migration-map?odoo_model=account.move&odoo_id=300", ALICE))["natural_key"] == "odoo:acme:account.move:300"
assert api("POST", "/migration/odoo/load", BUYER, {"cutover_date": "2026-06-30"})[0] == 403
def test_pur33_an_unmapped_supplier_is_listed_never_guessed(neighbours):
world()
odoo(neighbours)
neighbours.maps.partners.clear()
r = must(api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"}))
assert r["created"]["orders"] == 0 and any("not in business-partner's map" in s["reason"] for s in r["skipped"])
def test_pur34_reconcile_signs_only_when_equal_and_a_signed_run_is_frozen(neighbours):
world()
odoo(neighbours, qty_received="9") # Odoo says 9 received, the picking says 10
must(api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"}))
rec = must(api("POST", "/migration/odoo/odoo:acme:2026-06-30/reconcile", ALICE, {}))
bad = [c for c in rec["checks"] if not c["equal"]]
assert [c["check"] for c in bad] == ["received quantity per order line"] and rec["all_equal"] is False
s, b = api("POST", "/migration/odoo/odoo:acme:2026-06-30/sign", ALICE, {})
assert s == 409 and "differs" in str(b)
must(api("POST", "/migration/odoo/odoo:acme:2026-06-30/explain", ALICE, {"check_no": 2, "explanation": "qty_received was set by hand in Odoo"}))
assert must(api("POST", "/migration/odoo/odoo:acme:2026-06-30/sign", ALICE, {}))["status"] == "signed"
s, b = api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"})
assert s == 409 and "signed" in str(b)
def test_pur34_once_equal_a_run_signs(neighbours):
world()
odoo(neighbours)
must(api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"}))
rec = must(api("POST", "/migration/odoo/odoo:acme:2026-06-30/reconcile", ALICE, {}))
assert rec["all_equal"] is True and {"check", "odoo", "ours", "equal", "explanation"} <= set(rec["checks"][0])
sig = must(api("POST", "/migration/odoo/odoo:acme:2026-06-30/sign", ALICE, {}))
assert sig["status"] == "signed" and sig["signed_by_email"] == "alice@shop.example"
s, b = api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"})
assert s == 409 and "signed" in str(b)