dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_migration.py
"""INV-26 / the Odoo migration: opening balances at Odoo's cost, idempotent, history read-only, reconciled, signed runs frozen.
A fake Odoo and fake id maps; no live Odoo is ever read."""
import pytest
import world
from conftest import ACCOUNTANT, OPERATOR, SYSTEM, api, must, refused
from brick_inventory.infrastructure import odoo
class FakeOdoo:
def __init__(self):
self.data = {
"stock.warehouse": [{"id": 1, "name": "Leeds", "code": "WH", "company_id": 1, "active": True}],
"stock.location": [
{"id": 10, "name": "WH", "complete_name": "WH", "location_id": False, "usage": "view", "warehouse_id": 1, "active": True},
{"id": 11, "name": "Shelf 1", "complete_name": "WH/Stock/Shelf 1", "location_id": 10, "usage": "internal", "warehouse_id": 1, "active": True},
{"id": 12, "name": "Shelf 2", "complete_name": "WH/Stock/Shelf 2", "location_id": 10, "usage": "internal", "warehouse_id": 1, "active": True},
{"id": 99, "name": "Customers", "complete_name": "Partners/Customers", "location_id": False, "usage": "customer", "warehouse_id": False, "active": True}],
"stock.quant": [
{"id": 100, "product_id": 5, "location_id": 11, "lot_id": 50, "quantity": 8, "owner_id": False, "company_id": 1},
{"id": 101, "product_id": 6, "location_id": 12, "lot_id": False, "quantity": 120, "owner_id": False, "company_id": 1},
{"id": 102, "product_id": 6, "location_id": 99, "lot_id": False, "quantity": 5, "owner_id": False, "company_id": 1}],
"stock.lot": [{"id": 50, "name": "L-2611", "product_id": 5, "expiration_date": "2027-06-01"}],
"product.product": [{"id": 5, "default_code": "COLA-075", "standard_price": 21.60}, {"id": 6, "default_code": "SCREW-M4", "standard_price": 0.0245}],
"stock.valuation.layer": [{"product_id": 5, "company_id": 1, "quantity": 8, "value": 172.80}, {"product_id": 6, "company_id": 1, "quantity": 120, "value": 2.94}],
"stock.move": [{"id": 1, "reference": "WH/IN/00001", "state": "done", "origin": "P00001"}, {"id": 2, "reference": "WH/OUT/00001", "state": "done", "origin": "S00001"}],
"stock.move.line": [{"id": 1, "move_id": 1, "product_id": 5, "lot_id": 50, "quantity": 8, "location_id": 99, "location_dest_id": 11, "date": "2026-10-01 09:00:00"},
{"id": 2, "move_id": 2, "product_id": 6, "lot_id": False, "quantity": 5, "location_id": 12, "location_dest_id": 99, "date": "2026-10-02 09:00:00"}]}
def read(self, model, fields, domain=None):
return [dict(r) for r in self.data[model]]
class FakeMaps:
def product(self, odoo_id):
return {5: {"product_id": "COLA-075", "uom_numerator": 18, "uom_denominator": 1, "base_unit_code": "LTR"}, # Odoo counts packs of 18 L
6: {"product_id": "SCREW-M4", "uom_numerator": 1, "uom_denominator": 1, "base_unit_code": "C62"}}.get(odoo_id)
def partner(self, odoo_id):
return None
@pytest.fixture
def fake_odoo():
fo = FakeOdoo()
odoo.READER, odoo.MAPS = fo, FakeMaps()
yield fo
odoo.READER = odoo.MAPS = None
RUN = {"odoo_db": "shop", "cutover": "2026-11-01", "company_code": "UK1"}
def test_the_run_writes_opening_moves_at_odoo_cost_and_a_second_run_writes_nothing(fake_odoo):
world.catalogue()
first = must(api("POST", "/migration/odoo/import", SYSTEM, RUN))
assert first["created"]["opening_moves"] == 2 and first["created"]["warehouses"] == 1 and first["created"]["lots"] == 1 and first["created"]["history"] == 2
assert must(api("GET", "/stock/COLA-075", OPERATOR))["warehouses"][0]["on_hand"] == "144" # 8 packs of 18 L
assert must(api("GET", "/valuation/products/COLA-075", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "172.80"
assert must(api("GET", "/valuation/products/SCREW-M4", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "2.94"
again = must(api("POST", "/migration/odoo/import", SYSTEM, RUN))
assert again["unchanged"] is True and all(v == 0 for v in again["created"].values())
assert must(api("GET", "/stock/COLA-075", OPERATOR))["warehouses"][0]["on_hand"] == "144"
def test_reconcile_answers_every_check_and_history_is_read_only(fake_odoo):
world.catalogue()
must(api("POST", "/migration/odoo/import", SYSTEM, RUN))
rec = must(api("GET", "/migration/odoo/reconcile", SYSTEM, RUN))
assert rec["all_equal"] is True, [c for c in rec["checks"] if not c["equal"]]
assert {c["check"] for c in rec["checks"]} >= {"on hand per product per internal location (and lot)", "stock value per product per company", "history carried"}
assert all(set(c) == {"check", "odoo", "ours", "equal", "explanation"} for c in rec["checks"])
fake_odoo.data["stock.valuation.layer"][0]["value"] = 175.00
bad = must(api("GET", "/migration/odoo/reconcile", SYSTEM, RUN))
value = [c for c in bad["checks"] if c["check"].startswith("stock value")][0]
assert bad["all_equal"] is False and value["equal"] is False and "FIFO or standard" in value["explanation"]
hist = must(api("GET", "/migration/odoo/history", SYSTEM, {"odoo_db": "shop"}))
assert hist["count"] == 2
assert api("POST", "/migration/odoo/history", SYSTEM, {})[0] in (404, 405) # there is no write route
assert must(api("GET", "/stock/SCREW-M4", OPERATOR))["warehouses"][0]["on_hand"] == "120" # the 5 sent out in history do not count
def test_a_signed_run_is_frozen_and_negative_quants_stop_the_run(fake_odoo):
world.catalogue()
fake_odoo.data["stock.quant"].append({"id": 103, "product_id": 6, "location_id": 11, "lot_id": False, "quantity": -3, "owner_id": False, "company_id": 1})
body = refused(api("POST", "/migration/odoo/import", SYSTEM, RUN), 409, "negative quants")
assert body["detail"]["gate"] == "negative_quants"
ok = must(api("POST", "/migration/odoo/import", SYSTEM, {**RUN, "carry_negative": True}))
assert ok["negative_quants_carried"] == 1
assert must(api("POST", "/migration/odoo/sign", SYSTEM, RUN))["replay"] is False
assert must(api("POST", "/migration/odoo/sign", SYSTEM, RUN))["replay"] is True
frozen = refused(api("POST", "/migration/odoo/import", SYSTEM, {**RUN, "carry_negative": True}), 409, "frozen")
assert frozen["detail"]["gate"] == "run_signed"
refused(api("POST", "/migration/odoo/import", OPERATOR, RUN), 403, "inv:migration:import")