130 lines
5.6 KiB
Python
130 lines
5.6 KiB
Python
"""v4.9.12 tests for fiscal-customer Odoo fulfilment reconstruction.
|
|
|
|
The Odoo connector may only provide a reliable company name. Once a sale order
|
|
is staged, ClientFlow should still import order fulfilment evidence and rebuild
|
|
a useful process timeline: sale -> delivery done -> invoice to issue.
|
|
"""
|
|
|
|
import sys
|
|
import types
|
|
|
|
if "sqlalchemy" not in sys.modules:
|
|
sqlalchemy = types.ModuleType("sqlalchemy")
|
|
sqlalchemy.text = lambda sql: sql
|
|
sys.modules["sqlalchemy"] = sqlalchemy
|
|
|
|
if "app.db" not in sys.modules:
|
|
app_db = types.ModuleType("app.db")
|
|
app_db.engine = object()
|
|
sys.modules["app.db"] = app_db
|
|
|
|
if "app.config" not in sys.modules:
|
|
app_config = types.ModuleType("app.config")
|
|
app_config.settings = types.SimpleNamespace(odoo_enabled=True)
|
|
sys.modules["app.config"] = app_config
|
|
|
|
if "app.action_catalog" not in sys.modules:
|
|
action_catalog = types.ModuleType("app.action_catalog")
|
|
action_catalog.get_action_config = lambda code: {"route": "rever", "action": code, "action_required": True, "safe_to_post": False}
|
|
sys.modules["app.action_catalog"] = action_catalog
|
|
|
|
from app import external_reconciliation_sync as sync
|
|
from app import reconciliation_service as svc
|
|
|
|
|
|
class FakeOdooClient:
|
|
def search_read(self, model, domain=None, fields=None, *, limit=500, offset=0, order="id asc", context=None):
|
|
if model == "sale.order":
|
|
return [{
|
|
"id": 274,
|
|
"name": "S00274",
|
|
"state": "sale",
|
|
"date_order": "2026-06-02 11:25:49",
|
|
"partner_id": [288, "ELEGANTLEGACY, LDA"],
|
|
"amount_total": 537.0,
|
|
"currency_id": [125, "EUR"],
|
|
"invoice_status": "to invoice",
|
|
"order_line": [1001, 1002],
|
|
}]
|
|
if model == "res.partner":
|
|
return [{"id": 288, "name": "ELEGANTLEGACY, LDA", "vat": "", "email": ""}]
|
|
if model == "sale.order.line":
|
|
return [
|
|
{"id": 1001, "order_id": [274, "S00274"], "product_id": [11, "Wallbox 7.4KW"], "name": "Wallbox 7.4KW", "product_uom_qty": 1.0, "qty_delivered": 1.0, "qty_invoiced": 0.0, "price_unit": 179.0, "price_total": 179.0},
|
|
{"id": 1002, "order_id": [274, "S00274"], "product_id": [12, "Wallbox 11KW"], "name": "Wallbox 11KW", "product_uom_qty": 1.0, "qty_delivered": 1.0, "qty_invoiced": 0.0, "price_unit": 179.0, "price_total": 179.0},
|
|
]
|
|
if model == "stock.picking":
|
|
return [{"id": 295, "name": "WH/OUT/00295", "state": "done", "origin": "S00274", "picking_type_id": [1, "Delivery Orders"], "scheduled_date": "2026-06-02 12:25:00", "date_done": "2026-06-02 13:12:00"}]
|
|
if model == "mrp.production":
|
|
return []
|
|
return []
|
|
|
|
|
|
def test_odoo_sync_enriches_sale_with_lines_pickings_and_fulfilment(monkeypatch):
|
|
monkeypatch.setattr(sync.settings, "odoo_enabled", True, raising=False)
|
|
|
|
odoo_module = types.ModuleType("app.odoo_client")
|
|
odoo_module.OdooClient = lambda: FakeOdooClient()
|
|
monkeypatch.setitem(sys.modules, "app.odoo_client", odoo_module)
|
|
|
|
staged = []
|
|
monkeypatch.setattr(sync, "upsert_reconciliation_item", lambda **kw: staged.append(kw) or kw)
|
|
|
|
result = sync.sync_odoo_reconciliation_candidates(days=7, limit=10)
|
|
|
|
assert result["seen"] == 1
|
|
candidate = staged[0]
|
|
record = candidate["payload"]["record"]
|
|
assert record["partner_external_id"] == 288
|
|
assert record["order_lines"][0]["product_name"] == "Wallbox 7.4KW"
|
|
assert record["pickings"][0]["name"] == "WH/OUT/00295"
|
|
assert record["fulfilment"]["delivery_done"] is True
|
|
assert record["fulfilment"]["invoice_pending"] is True
|
|
assert candidate["suggested_action"] == "SEND_INVOICE"
|
|
|
|
|
|
def test_process_timeline_includes_delivery_done_and_invoice_pending():
|
|
item = {
|
|
"id": "odoo-sale-274",
|
|
"source_system": "odoo",
|
|
"external_type": "odoo_sale_order",
|
|
"document_number": "S00274",
|
|
"document_date": "2026-06-02",
|
|
"amount": "537.00",
|
|
"currency": "EUR",
|
|
"payload": {
|
|
"record": {
|
|
"name": "S00274",
|
|
"invoice_status": "to invoice",
|
|
"order_lines": [{"product_name": "Wallbox 7.4KW", "qty_delivered": 1.0, "qty_invoiced": 0.0}],
|
|
"fulfilment": {
|
|
"delivery_done": True,
|
|
"invoice_pending": True,
|
|
"physical_status": "shipped",
|
|
"outgoing_pickings": [{"name": "WH/OUT/00295", "state": "done", "date_done": "2026-06-02 13:12:00"}],
|
|
"lines": [{"product_name": "Wallbox 7.4KW", "qty_delivered": 1.0, "qty_invoiced": 0.0}],
|
|
},
|
|
}
|
|
},
|
|
}
|
|
|
|
state = svc.infer_reconciliation_process_state([item])
|
|
labels = [step["label"] for step in state["steps"]]
|
|
types = [step["external_type"] for step in state["steps"]]
|
|
|
|
assert "Venda/encomenda encontrada no Odoo" in labels
|
|
assert "Entrega Odoo concluída" in labels
|
|
assert "Fatura por emitir" in labels
|
|
assert "odoo_delivery" in types
|
|
assert "odoo_invoice_pending" in types
|
|
assert state["action_code"] == "SEND_INVOICE"
|
|
assert state["stage"] == "SHIPPED"
|
|
|
|
|
|
def test_static_hooks_for_partner_mapping_and_window_cleanup_reopen():
|
|
service_source = open("app/reconciliation_service.py", encoding="utf-8").read()
|
|
assert "external_customer_mappings" in service_source
|
|
assert "mapeamento Odoo confirmado" in service_source
|
|
assert "payload - 'window_cleanup'" in service_source
|
|
assert "_upsert_odoo_operation_links_from_item" in service_source
|