50 lines
2.3 KiB
Python
50 lines
2.3 KiB
Python
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def read(path: str) -> str:
|
|
return (ROOT / path).read_text(encoding="utf-8")
|
|
|
|
|
|
def test_v4923_reconciliation_sync_runs_customer_seed_before_documents():
|
|
sync = read("app/external_reconciliation_sync.py")
|
|
block = sync.split("async def sync_all_external_reconciliation_candidates", 1)[1]
|
|
|
|
assert "sync_external_fiscal_customers_for_reconciliation" in sync
|
|
assert "sync_jasmin_fiscal_customers_for_reconciliation" in sync
|
|
assert "sync_odoo_fiscal_customers_for_reconciliation" in sync
|
|
assert block.index("customer_result = await sync_external_fiscal_customers_for_reconciliation") < block.index("jasmin_result = await sync_jasmin_reconciliation_candidates")
|
|
assert block.index("jasmin_result = await sync_jasmin_reconciliation_candidates") < block.index("odoo_result = await asyncio.to_thread(sync_odoo_reconciliation_candidates")
|
|
assert "customer_seen" in block
|
|
assert "customers_created_or_updated" in block
|
|
|
|
|
|
def test_v4923_candidates_are_linked_to_seeded_fiscal_customer():
|
|
sync = read("app/external_reconciliation_sync.py")
|
|
|
|
assert "_upsert_fiscal_customer_from_external_record(record, source_system=\"jasmin\")" in sync
|
|
assert "_upsert_fiscal_customer_from_external_record(record, source_system=\"odoo\")" in sync
|
|
assert "\"customer_id\": customer_id" in sync
|
|
assert "source_system == \"odoo\" and not (tax_id or is_company or _looks_like_company_name(name))" in sync
|
|
|
|
|
|
def test_v4923_customer_upsert_uses_fiscal_name_when_nif_missing():
|
|
service = read("app/commercial_service.py")
|
|
|
|
assert "def normalize_fiscal_name" in service
|
|
assert "No domínio ClientFlow o nome fiscal de empresa é tratado como chave" in service
|
|
assert "normalized_name = normalize_fiscal_name" in service
|
|
assert "regexp_replace(lower(name)" in service
|
|
assert "metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB)" in service
|
|
|
|
|
|
def test_v4923_ui_explains_sequential_rebuild_pipeline():
|
|
page = read("app/admin_ui/pages/reconciliation.py")
|
|
script = read("scripts/sync_external_reconciliation.py")
|
|
|
|
assert "clientes fiscais → documentos Jasmin → vendas Odoo" in page
|
|
assert "clientes fiscais analisados" in page
|
|
assert "clientes criados/atualizados" in page
|
|
assert "--customers" in script
|