dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_migration.py
"""MFG-32: Odoo 17 mrp carried read-only (work centres, BOMs, routings, done orders as closed history with their lots, open orders decided at the
door), reconciled to Odoo per product and month, signed by the owner. A fake reader hands the rows; nothing is written in Odoo."""
import copy
import pytest
from conftest import ACCOUNTANT, OPERATOR, OWNER, PLANNER, QUALITY, SYSTEM, api, must, refused
from brick_manufacturing.infrastructure import odoo as odoo_port
DB = "roastery17"
class FakeOdoo:
def __init__(self):
self.wcs = [{"id": 1, "name": "Roaster 1", "code": "ROASTER-1", "costs_hour": 45, "capacity": 15},
{"id": 2, "name": "Packing bench", "code": "", "costs_hour": 22.5, "capacity": 100}]
self.bom_rows = [
{"id": 10, "product": "ROAST-ETH", "qty": "84", "uom": "KGM", "type": "normal", "code": "",
"lines": [{"product": "GREEN-ETH", "qty": "100", "uom": "KGM"}], "byproducts": [{"product": "CHAFF", "qty": "1.2", "uom": "KGM"}],
"operations": [{"id": 5, "name": "roast", "workcenter_id": 1, "sequence": 10, "time_cycle_manual": 20}]},
{"id": 11, "product": "LATTE-REG", "qty": "1", "uom": "C62", "type": "phantom", "code": "",
"lines": [{"product": "ROAST-HOUSE", "qty": "0.018", "uom": "KGM"}, {"product": "MILK-WHOLE", "qty": "0.25", "uom": "LTR"}],
"byproducts": [], "operations": []},
{"id": 12, "product": "BAG-ETH-250", "qty": "1", "uom": "C62", "type": "subcontract", "code": "", "lines": [], "byproducts": [], "operations": []}]
self.prods = [
{"id": 100, "name": "MO/0100", "state": "done", "product": "ROAST-ETH", "product_qty": "84", "qty_produced": "82.5", "uom": "KGM",
"date_finished": "2026-09-30 10:00:00", "lot_producing": "R-20260930-01",
"consumed": [{"product": "GREEN-ETH", "lot": "G-2609-01", "qty": "100"}], "workorders": [{"id": 1, "workcenter_id": 1, "sequence": 10, "duration": 160}]},
{"id": 101, "name": "MO/0101", "state": "done", "product": "ROAST-ETH", "product_qty": "84", "qty_produced": "84", "uom": "KGM",
"date_finished": "2026-10-15 09:00:00", "lot_producing": "R-20261015-01",
"consumed": [{"product": "GREEN-ETH", "lot": "G-2609-01", "qty": "100"}], "workorders": []},
{"id": 102, "name": "MO/0102", "state": "cancel", "product": "ROAST-ETH", "product_qty": "84", "qty_produced": "0", "uom": "KGM",
"date_finished": None, "lot_producing": "", "consumed": [], "workorders": []}]
def workcentres(self, db):
return copy.deepcopy(self.wcs)
def boms(self, db):
return copy.deepcopy(self.bom_rows)
def productions(self, db, years):
return copy.deepcopy(self.prods)
@pytest.fixture
def odoo():
reader = FakeOdoo()
odoo_port.ports.reader = reader
yield reader
odoo_port.ports.reader = None
BODY = {"odoo_db": DB, "cutover_date": "2026-11-16", "company_code": "UK01", "warehouse_id": "ROASTERY", "cost_centre": "ROASTERY",
"byproduct_nrv": {"CHAFF": "0.0000"}, "years": [2025, 2026]}
def load(body=None, who=None):
return api("POST", "/migration/odoo/load", who or SYSTEM, {**BODY, **(body or {})})
def test_mfg_32_work_centres_boms_routings_and_done_orders_are_carried(nb, odoo):
run = must(load(), 201)
assert run["run_id"] == f"mfg-odoo-{DB}-1" and run["status"] == "loaded"
assert run["summary"] == {"work_centres": 2, "boms": 1, "sale_boms": 1, "routings": 1, "subcontract_skipped": 1, "closed_orders": 2,
"open_orders": 0, "confirmations": 1}
wc = must(api("GET", "/work-centres/ROASTER-1", ACCOUNTANT))
assert wc["cost_centre"] == "ROASTERY" and wc["rates"] == [{"rate_kind": "machine", "valid_from": "2026-11-16", "rate_amount": "45.0000", "currency": "GBP"}]
assert must(api("GET", "/work-centres/Packing bench", ACCOUNTANT))["rates"][0]["rate_amount"] == "22.5000" # no code: the name, at the door
roast = must(api("GET", "/boms/ROAST-ETH:production", ACCOUNTANT))
v = roast["versions"][0]
assert (v["version"], v["status"], v["valid_from"], v["planned_yield_pct"]) == (1, "released", "2026-11-16", "84.0000")
assert [(x["kind"], x["product_id"], x["nrv_unit_cost"]) for x in v["lines"]] == [("component", "GREEN-ETH", "0.0000"), ("by_product", "CHAFF", "0.0000")]
assert must(api("GET", "/boms/LATTE-REG:sale", ACCOUNTANT))["versions"][0]["status"] == "released" # a phantom: a kit sold in the shops
assert must(api("GET", "/boms", ACCOUNTANT))["boms"] == ["LATTE-REG:sale", "ROAST-ETH:production"] # the subcontract BOM stays purchasing's
assert must(api("GET", "/routings/ROAST-ETH", ACCOUNTANT))["versions"][0]["operations"][0]["run_minutes"] == "20"
def test_mfg_32_a_done_order_is_a_read_only_closed_order_with_its_lot_and_minutes(nb, odoo):
must(load(), 201)
oid = f"odoo:{DB}:mrp.production:100"
got = must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))
assert (got["status"], got["product_id"], got["planned_quantity"], got["due_on"]) == ("closed", "ROAST-ETH", "84.0000", "2026-09-30")
assert [(m["kind"], m["lot_code"], m["quantity_base"], m["status"]) for m in got["movements"]] == [("output", "R-20260930-01", "82.5000", "sent")]
assert got["confirmations"][0]["machine_minutes"] == "160.0000" and got["confirmations"][0]["cost"] == "0.00" # no value: gl carries Odoo's entries
for act in ("issue", "release", "close", "cancel"): # history is read-only
assert api("POST", f"/production-orders/{oid}/{act}", PLANNER if act in ("release", "cancel") else ACCOUNTANT, {"location_id": "X", "lines": [], "reason": "x"})[0] in (403, 409, 422)
assert must(api("GET", "/trace/GREEN-ETH/G-2609-01", QUALITY, {"direction": "forward"}))["lots"][1]["lot_code"] in ("R-20260930-01", "R-20261015-01")
assert must(api("GET", f"/migration-map/{DB}/mrp.production", ACCOUNTANT))["map"]["100"] == oid
def test_mfg_32_a_second_load_writes_nothing_new(nb, odoo):
must(load(), 201)
again = must(load(), 201)
assert again["run_id"] == f"mfg-odoo-{DB}-2"
assert again["summary"] == {"work_centres": 0, "boms": 0, "sale_boms": 0, "routings": 0, "subcontract_skipped": 1, "closed_orders": 0, "open_orders": 0,
"confirmations": 0}
assert len(must(api("GET", "/production-orders", ACCOUNTANT))["production_orders"]) == 2
def test_mfg_32_open_orders_are_finished_in_odoo_by_default_or_recreated_at_the_door(nb, odoo):
odoo.prods.append({"id": 103, "name": "MO/0103", "state": "progress", "product": "ROAST-ETH", "product_qty": "84", "qty_produced": "0", "uom": "KGM",
"date_finished": None, "lot_producing": "", "consumed": [], "workorders": []})
refused(load(), 409, "MO/0103", gate="open_orders")
assert must(api("GET", "/boms", ACCOUNTANT))["boms"] == [] # all or nothing: the refused load wrote nothing
run = must(load({"open_orders": "recreate"}), 201)
assert run["summary"]["open_orders"] == 1
got = must(api("GET", f"/production-orders/odoo:{DB}:mrp.production:103", ACCOUNTANT))
assert (got["status"], got["bom_version"], got["planned_quantity"]) == ("created", 1, "84.0000")
def test_mfg_32_the_doors_are_the_owners_not_odoos(nb, odoo):
refused(load({"byproduct_nrv": {}}), 409, "net realisable value", gate="byproduct_nrv")
refused(load({"cost_centre": "NOPE"}), 409, "no cost centre", gate="cost_centre")
refused(load({"byproduct_nrv": {"CHAFF": "x"}}), 422)
odoo.bom_rows[0]["lines"][0]["qty"] = "1e30"
refused(load(), 422)
odoo.bom_rows[0]["lines"][0]["qty"] = "100"
odoo.bom_rows[0]["lines"][0]["product"] = "MYSTERY"
assert load()[0] == 404 # product answers: no such product
odoo_port.ports.reader = None
refused(load(), 409, "no Odoo connection")
def test_mfg_32_the_run_is_reconciled_per_product_and_month_then_signed_by_the_owner(nb, odoo):
run = must(load(), 201)["run_id"]
got = must(api("POST", f"/migration/odoo/{run}/reconcile", SYSTEM))
checks = {c["check"]: c for c in got["checks"]}
assert got["status"] == "reconciled" and all(c["equal"] for c in checks.values())
assert checks["output_per_product_and_month"]["odoo"] == {"ROAST-ETH|2026-09": "82.5000", "ROAST-ETH|2026-10": "84.0000"} == checks["output_per_product_and_month"]["ours"]
assert (checks["boms_carried"]["odoo"], checks["phantom_boms_mapped_to_usage_sale"]["ours"]) == (2, 1) # subcontract is not carried
refused(api("POST", f"/migration/odoo/{run}/reconcile", SYSTEM), 409, gate="state")
refused(api("POST", f"/migration/odoo/{run}/sign", ACCOUNTANT, {}), 403)
signed = must(api("POST", f"/migration/odoo/{run}/sign", OWNER, {}))
assert signed["status"] == "signed" and signed["signed_by"] == "owner@roastery.example" and signed["signed_on"] == "2026-11-10"
refused(api("POST", f"/migration/odoo/{run}/sign", OWNER, {}), 409, gate="state")
refused(api("POST", f"/migration/odoo/{run}/discard", OWNER, {"reason": "oops"}), 409, gate="state") # a signed run is frozen
def test_mfg_32_a_difference_blocks_the_signature_until_the_owner_explains_it(nb, odoo):
run = must(load(), 201)["run_id"]
odoo.prods.append({"id": 104, "name": "MO/0104", "state": "done", "product": "ROAST-ETH", "product_qty": "84", "qty_produced": "10", "uom": "KGM",
"date_finished": "2026-10-20 09:00:00", "lot_producing": "R-X", "consumed": [{"product": "GREEN-ETH", "lot": "G-X", "qty": "12"}],
"workorders": []}) # finished in Odoo after the load
got = must(api("POST", f"/migration/odoo/{run}/reconcile", SYSTEM))
bad = {c["check"] for c in got["checks"] if not c["equal"]}
assert bad == {"output_per_product_and_month", "genealogy_of_output_lots"}
refused(api("POST", f"/migration/odoo/{run}/sign", OWNER, {}), 409, "neither equal nor explained", gate="checks_reconciled")
must(api("POST", f"/migration/odoo/{run}/sign", OWNER, {"explanations": {"output_per_product_and_month": "MO/0104 was finished after the extract; loaded in the next run",
"genealogy_of_output_lots": "same order"}}))
def test_mfg_32_a_discarded_run_is_terminal_and_a_load_is_the_systems(nb, odoo):
run = must(load(), 201)["run_id"]
refused(api("POST", f"/migration/odoo/{run}/discard", PLANNER, {"reason": "x"}), 403)
refused(api("POST", f"/migration/odoo/{run}/discard", OWNER, {"reason": " "}), 422)
assert must(api("POST", f"/migration/odoo/{run}/discard", OWNER, {"reason": "wrong cut-over date"}))["status"] == "discarded"
refused(api("POST", f"/migration/odoo/{run}/reconcile", SYSTEM), 409, gate="state")
refused(load(who=PLANNER), 403)
refused(load(who=OPERATOR), 403)
refused(api("GET", f"/migration/odoo/{run}", OPERATOR), 403)
assert must(api("GET", f"/migration/odoo/{run}", ACCOUNTANT))["discard_reason"] == "wrong cut-over date"