dodil/erp-brick-sales-order
DODIL ERP brick: sales order
tests/test_migration.py
"""SO-24 / SC-SO-08: the one-off Odoo migration — mapped, idempotent, history read-only, reconciled — with a fake Odoo and
fake id maps. No live Odoo is ever read."""
from decimal import Decimal
import httpx
from conftest import ADMIN, ALICE, FULFILLER, REP, SYSTEM, api, must, refused
from sqlalchemy import func, select
from brick_sales_order.infrastructure import seams
from brick_sales_order.infrastructure.tables import (
so_delivery,
so_invoice_request,
so_line,
so_migration_map,
so_order,
)
D = Decimal
RUN = {"cut_off": "2026-11-20", "migration_owner": "owner@shop.example", "history_since": "2026-01-01", "user_map": {"7": "rep@shop.example"}}
OPEN_A, OPEN_B, OPEN_C = "odoo:1:S00003", "odoo:1:S00004", "odoo:1:S00005"
def run(**kw):
return api("POST", "/migration/odoo/import", SYSTEM, {**RUN, **kw})
def counts(engine):
with engine.begin() as c:
return {t.name: c.execute(select(func.count()).select_from(t)).scalar() for t in (so_order, so_line, so_delivery, so_invoice_request, so_migration_map)}
def mirror_invoicing(neighbours):
"""What ar's own migration leaves behind: invoice lines carrying sales_order_id and line_no."""
neighbours.invoicing.issued.update({"odoo:1:S00003": {1: D(5)}, "odoo:1:S00007": {1: D(1)}, "odoo:1:S00008": {1: D(2)}})
def test_SO_24_the_run_imports_open_orders_live_and_the_rest_as_history(engine):
r = must(run())
assert r["created"] == {"confirmed": 3, "done": 2, "cancelled": 1} and r["unchanged"] == 0 and r["in_scope"] == 6
assert r["quotations_left_to_brick_quote"] == 2 and r["outside_history_window"] == 1 # S00006 is 2024: not in a 2026 window
assert r["lines_not_carried"] == {"line_section": 1, "downpayment": 1, "line_note": 1}
a = must(api("GET", f"/sales-orders/{OPEN_A}", ADMIN))
assert (a["order_no"], a["lifecycle"], a["bp_id"], a["owner"], a["customer_po_ref"], a["currency"]) == ("S00003", "confirmed", "7311", "rep@shop.example", "PO-77", "GBP")
assert a["ordered_on"] == "2026-09-10" and a["confirmed_on"] == "2026-09-10" and a["revision"] == 1 and a["payment_terms_id"] == "NET30"
l1 = a["lines"][0]
assert (l1["product_id"], l1["unit_code"], D(l1["quantity"]), D(l1["unit_price"]), l1["priced_from"], l1["odoo_line_id"]) == ("WIDGET", "C62", 12, D("10.00"), "odoo:Retail GBP", "1001")
assert l1["priced_on"] == "2026-09-10" and l1["invoicing_policy"] == "delivered" and l1["tax_code"] == "S" and D(l1["tax_amount"]) == D("24.00")
assert len(a["lines"]) == 1 # the section line is not carried
assert D(l1["delivered"]) == 5 and a["deliveries"][0]["kind"] == "opening" and a["deliveries"][0]["delivered_on"] == "2026-11-20" \
and a["deliveries"][0]["delivery_ref"] == "odoo:1001" and a["deliveries"][0]["source"] == "migration"
assert a["fulfilment_status"] == "partly delivered"
def test_SO_24_prices_are_copied_never_repriced_and_a_unit_keeps_its_conversion():
must(run())
c = must(api("GET", f"/sales-orders/{OPEN_C}", ADMIN))["lines"][0]
assert (c["unit_code"], c["unit_numerator"], D(c["list_price"]), D(c["discount_pct"]), D(c["unit_price"]), D(c["net_amount"])) == ("PK24", 24, D("25.2631"), 5, D("23.9999"), D("72.00"))
assert D(must(api("GET", f"/sales-orders/{OPEN_C}", ADMIN))["total"]) == D("86.40")
neighbours_calls = None
assert neighbours_calls is None
def test_SO_24_a_down_payment_is_a_deposit_request_never_a_line_and_stays_unused(neighbours):
must(run())
b = must(api("GET", f"/sales-orders/{OPEN_B}", ADMIN))
assert [x["product_id"] for x in b["lines"]] == ["FEE"] and b["owner"] == "owner@shop.example" # an unknown salesperson: the migration owner
dep = b["invoice_requests"]
assert len(dep) == 1 and dep[0]["kind"] == "deposit" and D(dep[0]["amount"]) == D("5000.00") and dep[0]["invoice_id"] == "odoo:odoo17:901"
assert dep[0]["requested_by"] == "migration"
must(api("POST", f"/sales-orders/{OPEN_B}/request-invoice", ALICE, {})) # a later invoice still deducts what is unused
d = neighbours.invoicing.drafts[f"so:{OPEN_B}:2"]
assert [x["net_amount"] for x in d["lines"]] == ["10000.00"] and [(x["deposit_invoice_id"], x["net_amount"]) for x in d["deductions"]] == [("odoo:odoo17:901", "5000.00")]
def test_SO_24_the_second_run_changes_nothing(engine):
must(run())
before = counts(engine)
again = must(run())
assert again["created"] == {"confirmed": 0, "done": 0, "cancelled": 0} and again["unchanged"] == 6
assert counts(engine) == before and before["so_order"] == 6 and before["so_invoice_request"] == 1
assert before["so_migration_map"] == 6 + 6 + 3 # 6 orders, 6 carried lines, 3 not-carried lines
def test_SO_24_history_is_read_only_and_asks_invoicing_for_nothing(neighbours, engine):
must(run())
done, cancelled = "odoo:1:S00007", "odoo:1:S00009"
assert must(api("GET", f"/sales-orders/{done}", ADMIN))["lifecycle"] == "done" and must(api("GET", f"/sales-orders/{cancelled}", ADMIN))["lifecycle"] == "cancelled"
assert must(api("GET", "/sales-orders/odoo:1:S00008", ADMIN))["lifecycle"] == "done" # done and locked
for sid in (done, cancelled):
assert api("POST", f"/sales-orders/{sid}/amend", REP, {"reason": "x", "changes": [{"line_no": 1, "quantity": "2"}]})[0] in (403, 404)
assert "not_done" in refused(api("POST", f"/sales-orders/{sid}/amend", ALICE, {"reason": "x", "changes": [{"line_no": 1, "quantity": "2"}]}))
assert api("POST", f"/sales-orders/{sid}/cancel", ALICE, {"reason": "x"})[0] == (409 if sid == done else 200) # a cancelled one: already so
assert api("POST", f"/sales-orders/{sid}/deliveries", ALICE, {"line_no": 1, "quantity": "1", "delivery_ref": "X"})[0] == 409
assert api("POST", f"/sales-orders/{sid}/request-invoice", ALICE, {})[0] == 409
assert api("POST", f"/sales-orders/{sid}/lines", ALICE, {"line": {"product_id": "FEE", "quantity": "1"}})[0] == 409
assert neighbours.invoicing.drafts == {} and neighbours.invoicing.schedules == {}
with engine.begin() as c:
history = [r[0] for r in c.execute(select(so_order.c.sales_order_id).where(so_order.c.lifecycle != "confirmed"))]
assert c.execute(select(func.count()).select_from(so_invoice_request).where(so_invoice_request.c.sales_order_id.in_(history))).scalar() == 0
assert neighbours.tax.calls == [] # nothing re-taxed
def test_SO_24_odoo_numbers_are_kept_and_the_series_continues_after_them():
must(run())
assert must(api("GET", "/number-series/SO-2026", ALICE))["high_water_mark"] == 9 # S00009 is the highest of 2026
again = must(api("GET", "/number-series/SO-2026", ALICE))
assert again["high_water_mark"] == 9
o = must(api("POST", "/sales-orders", REP, {"bp_id": 7311, "currency": "GBP", "lines": [{"product_id": "FEE", "quantity": "1"}]}))
assert must(api("POST", f"/sales-orders/{o['sales_order_id']}/confirm", REP, {}))["order_no"] == "SO-2026-0010"
def test_SO_24_an_unmapped_record_on_an_open_order_stops_the_run_before_anything_is_written(neighbours, engine):
del neighbours.maps.products[203]
neighbours.maps.units.pop(2)
del neighbours.maps.partners[102]
said = refused(run())
assert "stopped before writing anything" in said and "partner: [102]" in said and "product: [203]" in said and "unit: [2]" in said
assert counts(engine)["so_order"] == 0
neighbours.maps.products[203] = {"product_id": "COLA-075", "invoicing_policy": "ordered"}
neighbours.maps.units[2] = {"unit_code": "PK24", "numerator": 24, "denominator": 1}
neighbours.maps.partners[102] = 7313
assert must(run())["created"]["confirmed"] == 3
def test_SO_24_a_dry_run_plans_and_writes_nothing(engine):
r = must(run(dry_run=True))
assert r["dry_run"] and r["in_scope"] == 6 and r["created"] == {"confirmed": 0, "done": 0, "cancelled": 0} and counts(engine)["so_order"] == 0
def test_SO_24_the_history_window_never_drops_an_open_order(neighbours):
neighbours.odoo._orders[2]["date_order"] = "2023-01-01 09:00:00" # S00003 is open and old
r = must(run())
assert r["created"]["confirmed"] == 3 and r["outside_history_window"] == 1
everything = must(run(history_since=None))
assert everything["in_scope"] == 7 and everything["created"]["done"] == 1
def test_SC_SO_08_the_reconciliation_has_zero_unexplained_differences(neighbours):
must(run())
mirror_invoicing(neighbours)
r = must(api("GET", "/migration/odoo/reconcile", SYSTEM, {"history_since": "2026-01-01"}))
by = {c["check"]: c for c in r["checks"]}
assert r["unexplained"] == [] and r["signable"] and set(by) == {
"orders per state", "draft and sent orders accounted for", "totals per currency, open and history", "lines carried", "lines not carried",
"prices unchanged", "to deliver per open line", "to invoice per open line", "invoiced per line", "unused deposits per open order"}
assert all(set(c) >= {"check", "odoo", "ours", "equal", "explanation"} for c in r["checks"])
assert by["orders per state"]["odoo"] == {"open": 3, "finished": 2, "cancelled": 1} == by["orders per state"]["ours"] and by["orders per state"]["equal"]
assert not by["draft and sent orders accounted for"]["equal"] and "brick-quote" in by["draft and sent orders accounted for"]["explanation"]
assert by["lines carried"]["equal"] and by["lines carried"]["odoo"] == 6
nc = by["lines not carried"]
assert nc["equal"] and nc["ours"] == {"line_section": 1, "downpayment": 1, "line_note": 1} and "section" in nc["explanation"]
tot = by["totals per currency, open and history"]
assert tot["equal"] and tot["odoo"]["GBP"]["untaxed"] == "19192.00" and "down-payment" in tot["explanation"]
assert by["to deliver per open line"]["odoo"] == by["to deliver per open line"]["ours"] == "20"
assert by["unused deposits per open order"]["odoo"] == "5000.00" and by["unused deposits per open order"]["equal"]
assert by["invoiced per line"]["equal"] and by["to invoice per open line"]["equal"] and by["prices unchanged"]["equal"]
def test_SC_SO_08_a_difference_with_no_reason_is_unexplained_and_blocks_the_signature(neighbours):
must(run())
mirror_invoicing(neighbours)
neighbours.invoicing.issued["odoo:1:S00003"] = {1: D(4)} # invoicing disagrees with Odoo
r = must(api("GET", "/migration/odoo/reconcile", SYSTEM, {"history_since": "2026-01-01"}))
assert not r["signable"] and set(r["unexplained"]) == {"invoiced per line", "to invoice per open line"}
assert "odoo:1:S00003 line 1: odoo 5, ours 4" in next(c for c in r["checks"] if c["check"] == "invoiced per line")["details"]
def test_SC_SO_08_without_invoicing_the_billed_checks_cannot_be_signed(monkeypatch):
must(run())
monkeypatch.setattr(seams.Ports, "invoicing_installed", lambda self: False)
r = must(api("GET", "/migration/odoo/reconcile", SYSTEM, {"history_since": "2026-01-01"}))
assert not r["signable"] and set(r["unexplained"]) == {"invoiced per line", "to invoice per open line"}
def test_the_migration_is_a_machine_only_and_not_a_persons():
assert api("POST", "/migration/odoo/import", ADMIN, {**RUN})[0] == 403 and api("POST", "/migration/odoo/import", FULFILLER, {**RUN})[0] == 403
assert api("POST", "/migration/odoo/import", SYSTEM, {"migration_owner": "x"})[0] == 422
assert "cut_off" in refused(api("POST", "/migration/odoo/import", SYSTEM, {**RUN, "cut_off": ""}), 422)
def test_the_odoo_reader_is_read_only_and_flattens_what_it_reads():
seen = []
def handler(request):
body = __import__("json").loads(request.content)
p = body["params"]
seen.append((p["service"], p["method"], p["args"][3:5] if p["service"] == "object" else None))
if p["service"] == "common":
return httpx.Response(200, json={"result": 2})
model = p["args"][3]
if model == "sale.order":
return httpx.Response(200, json={"result": [{"id": 1, "name": "S00001", "company_id": [1, "Co"], "state": "sale", "locked": False,
"partner_id": [101, "Acme"], "client_order_ref": False, "date_order": "2026-09-10 10:00:00",
"user_id": False, "currency_id": [3, "GBP"], "pricelist_id": [1, "Retail GBP"],
"payment_term_id": False, "incoterm": False, "amount_untaxed": 1, "amount_tax": 0, "amount_total": 1}]})
return httpx.Response(200, json={"result": [{"id": 9, "order_id": [1, "S00001"], "product_id": [201, "W"], "product_uom": [1, "Units"], "tax_id": [301]}]})
rd = seams.OdooJsonRpc("https://odoo.example", "db", "me", "key", transport=httpx.MockTransport(handler))
o = rd.orders()[0]
assert (o["company_id"], o["partner_id"], o["currency"], o["pricelist_name"], o["user_id"], o["payment_term_id"]) == (1, 101, "GBP", "Retail GBP", None, None)
ln = rd.lines([1])[0]
assert (ln["order_id"], ln["product_id"], ln["product_uom"], ln["tax_ids"]) == (1, 201, 1, [301])
assert {m for _, m, _ in seen} <= {"login", "execute_kw"} and all(x[1] != "write" for x in seen)
assert [x[2][1] for x in seen if x[0] == "object"] == ["search_read", "search_read"]
def test_later_bricks_read_the_odoo_map_through_one_route():
"""shared.yaml conventions.odoo_migration: GET /migration-map/{db}/{model}?ids= -> [{odoo_model, odoo_id, ours}];
a sale.order.line is "<sales_order_id>|<line_no>" (what ar's invoice lines name); an unmapped id is left out."""
must(run())
got = must(api("GET", "/migration-map/odoo17/sale.order.line", ADMIN, {"ids": "1001,999999"}))
assert got == [{"odoo_model": "sale.order.line", "odoo_id": "1001", "ours": f"{OPEN_A}|1"}]
assert must(api("GET", "/migration-map/odoo17/sale.order", ADMIN, {"ids": "3"}))[0]["ours"] == OPEN_A