Files
clientflow_backend/tests/test_v499_process_timeline_reconstruction_behavior.py

309 lines
12 KiB
Python

"""Behavior tests for v4.9.9 process timeline reconstruction.
These tests validate the core reconstruction rules without touching the real
production database: grouping is based on strong identity signals and the
suggested stage/action is inferred from the most advanced external evidence.
"""
import sys
import types
# The project test suite is intentionally lightweight and mostly static. This
# file exercises pure reconstruction helpers without requiring a live DB or
# SQLAlchemy in the test environment.
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.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 reconciliation_service as svc
def test_process_state_reconstructs_quote_then_odoo_sale_as_invoice_check():
state = svc.infer_reconciliation_process_state([
{
"id": "quote-1",
"source_system": "jasmin",
"external_type": "jasmin_quotation",
"document_number": "ORC.ORC2026.157",
"document_date": "2026-06-05",
"amount": "320.00",
},
{
"id": "sale-1",
"source_system": "odoo",
"external_type": "odoo_sale_order",
"document_number": "S00275",
"document_date": "2026-06-05",
"amount": "320.00",
},
])
assert state["stage"] == "ODOO_ORDER_CREATED"
assert state["action_code"] == "SEND_INVOICE"
assert [step["external_type"] for step in state["steps"]] == [
"jasmin_quotation",
"odoo_sale_order",
]
def test_process_state_payment_proof_requires_manual_payment_confirmation():
state = svc.infer_reconciliation_process_state([
{
"id": "invoice-1",
"source_system": "jasmin",
"external_type": "jasmin_invoice",
"document_number": "FA.FA2026.209",
"document_date": "2026-06-05",
"amount": "537.00",
},
{
"id": "proof-1",
"source_system": "manual",
"external_type": "payment_proof",
"document_number": "comprovativo.jpg",
"document_date": "2026-06-05",
"amount": "537.00",
},
])
assert state["stage"] == "INVOICE_SENT"
assert state["action_code"] == "CONFIRM_PAYMENT"
assert any(step["external_type"] == "payment_proof" for step in state["steps"])
def test_process_candidates_group_by_exact_nif_and_ignore_unrelated_lone_items(monkeypatch):
monkeypatch.setattr(
svc,
"list_reconciliation_items",
lambda *, status, limit, days: [
{
"id": "quote-1",
"source_system": "jasmin",
"external_type": "jasmin_quotation",
"title": "Orçamento Jasmin sem oportunidade · ORC.ORC2026.157",
"customer_name": "DOCTOR SOLUTION CONSTRUÇÕES, UNIPESSOAL, LDA",
"customer_tax_id": "PT515708690",
"document_number": "ORC.ORC2026.157",
"document_date": "2026-06-05",
"amount": "320.00",
"currency": "EUR",
},
{
"id": "sale-1",
"source_system": "odoo",
"external_type": "odoo_sale_order",
"title": "Venda Odoo sem oportunidade · S00275",
"customer_name": "DOCTOR SOLUTION CONSTRUÇÕES, UNIPESSOAL, LDA",
"customer_tax_id": "515708690",
"document_number": "S00275",
"document_date": "2026-06-05",
"amount": "320.00",
"currency": "EUR",
},
{
"id": "lone-1",
"source_system": "jasmin",
"external_type": "jasmin_quotation",
"title": "Orçamento isolado",
"customer_name": "CLIENTE SEM SEGUNDA EVIDÊNCIA",
"customer_tax_id": "599999999",
"document_number": "ORC.ISOLATED",
"document_date": "2026-06-05",
"amount": "10.00",
"currency": "EUR",
},
],
)
candidates = svc.list_reconciliation_process_candidates(status="open", days=3, limit=10)
assert len(candidates) == 1
candidate = candidates[0]
assert candidate["match_key"] == "nif"
assert candidate["match_value"] == "515708690"
assert candidate["confidence"] == "alta"
assert candidate["item_ids"] == ["quote-1", "sale-1"]
assert candidate["suggested_stage"] == "ODOO_ORDER_CREATED"
assert candidate["suggested_action"] == "SEND_INVOICE"
def test_single_item_with_open_operation_suggestion_becomes_process_candidate(monkeypatch):
opportunity_id = "11111111-1111-1111-1111-111111111111"
monkeypatch.setattr(
svc,
"list_reconciliation_items",
lambda *, status, limit, days: [
{
"id": "invoice-1",
"source_system": "jasmin",
"external_type": "jasmin_invoice",
"title": "Fatura Jasmin sem oportunidade · FA.FA2026.209",
"customer_name": "ELEGANTLEGACY, LDA",
"customer_tax_id": "515510165",
"document_number": "FA.FA2026.209",
"document_date": "2026-06-05",
"amount": "537.00",
"currency": "EUR",
"operation_suggestions": [
{
"opportunity_id": opportunity_id,
"title": "Elegantlegacy Lda · Emitir fatura",
"reason": "NIF exato",
}
],
}
],
)
candidates = svc.list_reconciliation_process_candidates(status="open", days=3, limit=10)
assert len(candidates) == 1
assert candidates[0]["item_ids"] == ["invoice-1"]
assert candidates[0]["suggestions"][0]["opportunity_id"] == opportunity_id
assert candidates[0]["suggested_action"] == "CONFIRM_PAYMENT"
def test_process_candidates_merge_jasmin_nif_with_odoo_name_only_evidence(monkeypatch):
"""Jasmin may provide NIF while Odoo only provides the fiscal name.
The real operational process should still be shown as one candidate, not as
an isolated Jasmin quote with an Odoo sale merely suggested elsewhere.
"""
monkeypatch.setattr(
svc,
"list_reconciliation_items",
lambda *, status, limit, days: [
{
"id": "quote-aczco",
"source_system": "jasmin",
"external_type": "jasmin_quotation",
"title": "Orçamento Jasmin sem oportunidade · ORC.ORC2026.154",
"customer_name": "ACZCO BRAGA ENERGY, LDA",
"customer_tax_id": "517249200",
"document_number": "ORC.ORC2026.154",
"document_date": "2026-06-03",
"amount": None,
"currency": "EUR",
"operation_suggestions": [],
},
{
"id": "sale-aczco",
"source_system": "odoo",
"external_type": "odoo_sale_order",
"title": "Venda Odoo sem oportunidade · S00279",
"customer_name": "ACZCO BRAGA ENERGY, LDA",
"customer_tax_id": "",
"document_number": "S00279",
"document_date": "2026-06-05",
"amount": "638.60",
"currency": "EUR",
"payload": {
"record": {
"invoice_status": "to invoice",
"fulfilment": {
"invoice_pending": True,
"lines": [
{"product_name": "Wallbox 7.4KW"},
{"product_name": "RFID Reader"},
],
"outgoing_pickings": [{"name": "WH/OUT/00300", "state": "done", "date_done": "2026-06-05 15:12:00"}],
},
}
},
"operation_suggestions": [],
},
],
)
candidates = svc.list_reconciliation_process_candidates(status="open", days=7, limit=10)
assert len(candidates) == 1
candidate = candidates[0]
assert candidate["match_key"] == "nif"
assert candidate["match_value"] == "517249200"
assert candidate["confidence"] == "alta"
assert candidate["identity_reason"] == "NIF + nome fiscal"
assert set(candidate["item_ids"]) == {"quote-aczco", "sale-aczco"}
assert candidate["suggested_stage"] == "SHIPPED"
assert candidate["suggested_action"] == "SEND_INVOICE"
assert any(step["external_type"] == "jasmin_quotation" for step in candidate["timeline"])
assert any(step["external_type"] == "odoo_sale_order" for step in candidate["timeline"])
assert any(step["external_type"] == "odoo_delivery" for step in candidate["timeline"])
def test_process_candidates_bridge_jasmin_nif_to_odoo_name_with_location_suffix(monkeypatch):
"""Odoo may only expose a display name with address/location suffixes.
Jasmin remains the fiscal identity source because it has the NIF. Odoo
evidence should still join the same reconstructed process when its company
name is a strong normalized name match.
"""
monkeypatch.setattr(
svc,
"list_reconciliation_items",
lambda *, status, limit, days: [
{
"id": "quote-aczco-fiscal",
"source_system": "jasmin",
"external_type": "jasmin_quotation",
"title": "Orçamento Jasmin sem oportunidade · ORC.ORC2026.154",
"customer_name": "ACZCO BRAGA ENERGY, LDA",
"customer_tax_id": "517249200",
"document_number": "ORC.ORC2026.154",
"document_date": "2026-06-03",
"amount": None,
"currency": "EUR",
"operation_suggestions": [],
},
{
"id": "sale-aczco-display-name",
"source_system": "odoo",
"external_type": "odoo_sale_order",
"title": "Venda Odoo sem oportunidade · S00279",
"customer_name": "ACZCO BRAGA ENERGY, LDA MERELIM SÃO PAIO Portugal",
"customer_tax_id": "",
"document_number": "S00279",
"document_date": "2026-06-05",
"amount": "638.60",
"currency": "EUR",
"payload": {
"record": {
"invoice_status": "to invoice",
"fulfilment": {
"invoice_pending": True,
"lines": [{"product_name": "Wallbox 11KW"}],
"outgoing_pickings": [{"name": "WH/OUT/00301", "state": "done", "date_done": "2026-06-05 15:12:00"}],
},
}
},
"operation_suggestions": [],
},
],
)
candidates = svc.list_reconciliation_process_candidates(status="open", days=7, limit=10)
assert len(candidates) == 1
candidate = candidates[0]
assert candidate["match_key"] == "nif"
assert candidate["match_value"] == "517249200"
assert candidate["confidence"] == "alta"
assert candidate["identity_reason"] == "NIF + nome fiscal"
assert set(candidate["item_ids"]) == {"quote-aczco-fiscal", "sale-aczco-display-name"}
assert candidate["suggested_stage"] == "SHIPPED"
assert candidate["suggested_action"] == "SEND_INVOICE"
assert any(step["external_type"] == "jasmin_quotation" for step in candidate["timeline"])
assert any(step["external_type"] == "odoo_sale_order" for step in candidate["timeline"])
assert any(step["external_type"] == "odoo_delivery" for step in candidate["timeline"])