""" Invoice pipeline end-to-end test runner. First run creates: test_data/master_data.xlsx — vendor, bank, tax, entity, currency master test_data/invoice_inputs.xlsx — 17 test cases (one row each) Subsequent runs read from those files — edit them freely. Results are always written to pipeline_test_results.xlsx. Usage: cd "Matching Functions" python run_pipeline_tests.py """ from __future__ import annotations import os import re import sys import traceback import unicodedata from datetime import date, datetime from decimal import Decimal from types import SimpleNamespace HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) # ── Pipeline imports — no logic rewritten here ──────────────────────────────── from invoice_pipeline import run_invoice_pipeline from two_way_match import InvoiceHeader, InvoiceLine, POHeader, POLine from three_way_match import GRLineItem from contract_match import ContractInvoiceHeader, ContractRecord, RateCardItem try: import openpyxl from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter except ImportError: sys.exit("openpyxl required. Run: pip install openpyxl") # ── Paths ───────────────────────────────────────────────────────────────────── DATA_DIR = os.path.join(HERE, "test_data") MASTER_FILE = os.path.join(DATA_DIR, "master_data.xlsx") INPUT_FILE = os.path.join(DATA_DIR, "invoice_inputs.xlsx") RESULTS_FILE = os.path.join(HERE, "pipeline_test_results.xlsx") os.makedirs(DATA_DIR, exist_ok=True) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 1 — SEED DATA # ══════════════════════════════════════════════════════════════════════════════ _MASTER_SHEETS: dict = { "Vendors": { "headers": [ "id", "tenant_id", "legal_name", "vendor_name", "vendor_status", "deleted_at", "billing_country", "billing_state", "tax_registration_number", "payment_terms", "exemption_status", ], "rows": [ [100, 1, "Acme Corporation Ltd", "Acme Corporation Ltd", "active", None, "GB", None, "GB123456715", 30, "standard"], ], }, "Vendor_Tax_IDs": { "headers": [ "tenant_id", "vendor_id", "tax_id_canonical", "tax_id_type", "verified_at", "deleted_at", "effective_from", "effective_to", ], "rows": [ [1, 100, "123456715", "VAT-GB", "2024-01-01", None, None, None], ], }, "Vendor_Bank_Accounts": { "headers": [ "tenant_id", "vendor_id", "iban_canonical", "acc_no_canonical", "swift_bic", "bank_acc_name", "currency_code", "bank_country", "is_primary", "effective_from", "effective_to", "deleted_at", ], "rows": [ [1, 100, "GB29NWBK60161331926819", "31926819", "NWBKGB2L", "Acme Corporation Ltd", "GBP", "GB", True, None, None, None], ], }, "Entity": { "headers": ["id", "tenant_id", "entity_id", "country_code", "vat_id", "region_code"], "rows": [ [1, 1, 10, "GB", "GB987654321", None], ], }, "Tax_Master": { "headers": [ "id", "tenant_id", "is_active", "deleted_at", "country_code", "region_code", "tax_type", "tax_rate", "tax_name", "effective_from", "effective_to", ], "rows": [ [1, 1, True, None, "GB", None, "VAT", 20.0, "VAT Standard 20%", None, None], [2, 1, True, None, "GB", None, "VAT", 5.0, "VAT Reduced 5%", None, None], ], }, "Currency": { "headers": ["currency_code", "minor_unit_value"], "rows": [ ["GBP", "0.01"], ["USD", "0.01"], ["EUR", "0.01"], ], }, } # ── Invoice inputs column order ─────────────────────────────────────────────── _INPUT_HEADERS = [ # metadata "tc_id", "scenario", "details", "expected", # document "doc_id", "tenant_id", "entity_id", "doc_status", "doc_type", "tenant_timezone", # extracted — vendor identity "sender_name", "sender_name_canonical", "sender_tax_id", "sender_tax_id_canonical", # extracted — bank "bank_iban", "bank_iban_canonical", "bank_acc_no", "bank_swift", "bank_acc_name", # extracted — dates "invoice_date", "due_date", "due_date_origin", "service_start_date", "service_end_date", # extracted — terms / PO ref "payment_terms", "po_number_canonical", # extracted — amounts "currency", "invoice_no", "total_amount", "subtotal", "tax_amount", "tax_rate", # single line item "line1_description", "line1_sku", "line1_quantity", "line1_unit_price", "line1_amount", "line1_tax_rate", "line1_tax_amount", # validation purchase-orders (payment-terms check) "val_po_number", "val_po_payment_terms", # matching controls "has_po", "po_status", "po_line_status", "has_gr", "use_contract", # expected checks "exp_resolved_vendor_id", "exp_route", "exp_risk_flags_present", "exp_risk_flags_absent", "exp_match_exception_type", ] # ── Base row shared by all test cases ───────────────────────────────────────── _B: dict = { "doc_id": 1001, "tenant_id": 1, "entity_id": 10, "doc_status": "validated", "doc_type": "invoice", "tenant_timezone": "Europe/London", "sender_name": "Acme Corporation Ltd", "sender_name_canonical": "ACMECORPORATIONLTD", "sender_tax_id": "GB123456715", "sender_tax_id_canonical": "GB123456715", "bank_iban": "GB29NWBK60161331926819", "bank_iban_canonical": "GB29NWBK60161331926819", "bank_acc_no": "31926819", "bank_swift": "NWBKGB2L", "bank_acc_name": "Acme Corporation Ltd", "invoice_date": "2025-10-01", "due_date": "2025-10-31", "due_date_origin": "invoice", "service_start_date": None, "service_end_date": None, "payment_terms": 30, "po_number_canonical": None, "currency": "GBP", "invoice_no": "INV-BASE", "total_amount": "120.00", "subtotal": "100.00", "tax_amount": "20.00", "tax_rate": "20.0", "line1_description": "Software License - Annual", "line1_sku": "SW-LIC-001", "line1_quantity": "10", "line1_unit_price": "10.00", "line1_amount": "100.00", "line1_tax_rate": "20.0", "line1_tax_amount": "20.00", "val_po_number": None, "val_po_payment_terms": None, "has_po": False, "po_status": "open", "po_line_status": "open", "has_gr": False, "use_contract": False, "exp_resolved_vendor_id": 100, "exp_route": None, "exp_risk_flags_present": None, "exp_risk_flags_absent": None, "exp_match_exception_type": None, } # Shared PO overrides _PO = { "po_number_canonical": "PO-001", "has_po": True, "val_po_number": "PO-001", "val_po_payment_terms": 30, } def _tc(tc_id: str, scenario: str, details: str, expected: str, **kw) -> dict: row = {**_B, **kw, "tc_id": tc_id, "scenario": scenario, "details": details, "expected": expected} if "invoice_no" not in kw: row["invoice_no"] = f"INV-{tc_id}" return row _TEST_CASES_SEED = [ _tc("TC01", "2-way match – clean PO invoice", "All validation checks pass. PO open, no GR.", "vendor=100, route=two_way, no risk flags", **_PO, exp_route="two_way", exp_risk_flags_absent="unknown_vendor,tax_id_mismatch,bank_iban_mismatch", ), _tc("TC02", "3-way match – PO + GR", "Same as TC01 but a goods-receipt row exists for the PO line.", "vendor=100, route=three_way, no risk flags", **_PO, has_gr=True, exp_route="three_way", exp_risk_flags_absent="unknown_vendor", ), _tc("TC03", "Contract match – no PO on invoice", "Vendor resolved. No PO reference. Rate card line matches invoice line exactly.", "vendor=100, route=contract, no risk flags", use_contract=True, exp_route="contract", exp_risk_flags_absent="unknown_vendor", ), _tc("TC04", "Unknown vendor – stub (cannot match)", "Sender name and tax ID don't match any vendor. No PO, no contract.", "route=stub, risk=unknown_vendor, exception=cannot_match_no_vendor", sender_name="Unknown Vendor Corp", sender_name_canonical="UNKNOWNVENDORCORP", sender_tax_id="XX999999999", sender_tax_id_canonical="XX999999999", exp_resolved_vendor_id=None, exp_route="stub", exp_risk_flags_present="unknown_vendor", exp_match_exception_type="cannot_match_no_vendor", ), _tc("TC05", "Tax ID mismatch (vendor resolved by name)", "Sender name matches Acme Corp but invoice tax ID is wrong. Vendor still resolved.", "vendor=100, route=two_way, risk=tax_id_mismatch", **_PO, sender_tax_id="GB000000000", sender_tax_id_canonical="GB000000000", exp_route="two_way", exp_risk_flags_present="tax_id_mismatch", ), _tc("TC06", "Bank IBAN mismatch", "Vendor and tax ID OK but IBAN on invoice does not match master record.", "vendor=100, route=two_way, risk=bank_iban_mismatch", **_PO, bank_iban="GB00WRONG0000000000001", bank_iban_canonical="GB00WRONG0000000000001", exp_route="two_way", exp_risk_flags_present="bank_iban_mismatch", ), _tc("TC07", "Invoice date too old (> max_invoice_age_days=365)", "Invoice date 2020-01-01 exceeds the 365-day age limit.", "route=two_way, risk=date_anomaly_invoice_too_old", **_PO, invoice_date="2020-01-01", due_date="2020-01-31", exp_route="two_way", exp_risk_flags_present="date_anomaly_invoice_too_old", ), _tc("TC07a", "Future invoice date", "Invoice date 2030-01-01 is in the future.", "route=two_way, risk=date_anomaly_invoice_future_dated", **_PO, invoice_date="2030-01-01", due_date="2030-01-31", exp_route="two_way", exp_risk_flags_present="date_anomaly_invoice_future_dated", ), _tc("TC07b", "Service period reversed (end before start)", "service_end_date (Oct 1) is earlier than service_start_date (Oct 31).", "route=two_way, risk=date_anomaly_service_period_reversed", **_PO, service_start_date="2025-10-31", service_end_date="2025-10-01", exp_route="two_way", exp_risk_flags_present="date_anomaly_service_period_reversed", ), _tc("TC08", "Payment terms mismatch (invoice=60 days, PO=30 days)", "Invoice declares 60-day terms but the PO specifies 30 days.", "route=two_way, risk=payment_terms_mismatch", **_PO, payment_terms=60, exp_route="two_way", exp_risk_flags_present="payment_terms_mismatch", ), _tc("TC09", "Missing invoice number (incomplete_fields)", "invoice_no is absent. Completeness check flags incomplete_fields.", "route=two_way, risk=incomplete_fields", **_PO, invoice_no=None, exp_route="two_way", exp_risk_flags_present="incomplete_fields", ), _tc("TC09a", "Tax rate not in master (line rate=25%, master has 20% and 5% only)", "Line declares 25% tax rate. GB tax master has only 20% and 5%.", "route=two_way, risk=tax_rate_not_in_master", **_PO, tax_rate="25.0", tax_amount="25.00", total_amount="125.00", line1_tax_rate="25.0", line1_tax_amount="25.00", exp_route="two_way", exp_risk_flags_present="tax_rate_not_in_master", ), _tc("TC09b", "Math error: line sum (80.00) != subtotal (100.00)", "Single line amount is 80.00 but invoice subtotal is 100.00.", "route=two_way, risk=math_error_line_items", **_PO, line1_amount="80.00", line1_tax_amount="16.00", # 80.00 x 20% = 16.00 (consistent with line amount) exp_route="two_way", exp_risk_flags_present="math_error_line_items", ), _tc("TC10", "PO not found (PO number on invoice but no DB match)", "Invoice references PO-NOT-FOUND which does not exist in the system.", "route=stub, exception=po_not_found", po_number_canonical="PO-NOT-FOUND", has_po=False, exp_route="stub", exp_match_exception_type="po_not_found", ), _tc("TC11", "PO cancelled", "Invoice references PO-001 which is in cancelled status.", "route=stub, exception=po_cancelled", **_PO, po_status="cancelled", exp_route="stub", exp_match_exception_type="po_cancelled", ), _tc("TC12", "Price variance – red zone (50% above PO price for services)", "Invoice unit price 15.00 vs PO unit price 10.00 (50% variance; red threshold >15%).", "route=two_way, match result in red zone", **_PO, line1_unit_price="15.00", line1_amount="150.00", line1_tax_amount="30.00", # 150.00 x 20% = 30.00 subtotal="150.00", tax_amount="30.00", total_amount="180.00", exp_route="two_way", ), _tc("TC13", "Contract match – exact rate card price", "No PO. Vendor resolved. Invoice line price matches rate card exactly (10.00).", "route=contract, overall=matched or partial_match", use_contract=True, exp_route="contract", ), _tc("TC14", "Contract match – amber zone (12% above rate card)", "Invoice unit price 11.20 is 12% above rate card 10.00. Services amber band is 10-15%.", "route=contract, amber zone", use_contract=True, line1_unit_price="11.20", line1_amount="112.00", line1_tax_amount="22.40", # 112.00 x 20% = 22.40 subtotal="112.00", tax_amount="22.40", total_amount="134.40", exp_route="contract", ), _tc("TC15", "PO closed with all lines closed", "PO-001 status=closed and PO line status=closed. Router stubs with po_closed.", "route=stub, exception=po_closed", **_PO, po_status="closed", po_line_status="closed", exp_route="stub", exp_match_exception_type="po_closed", ), ] # ══════════════════════════════════════════════════════════════════════════════ # SECTION 2 — EXCEL CREATORS # ══════════════════════════════════════════════════════════════════════════════ _HDR_FILL = PatternFill("solid", fgColor="1F4E79") _HDR_FONT = Font(bold=True, color="FFFFFF", size=10) _HDR_ALIGN = Alignment(horizontal="center", wrap_text=True) def _write_headers(ws, headers: list): ws.append(headers) for col, _ in enumerate(headers, 1): cell = ws.cell(row=1, column=col) cell.fill = _HDR_FILL cell.font = _HDR_FONT cell.alignment = _HDR_ALIGN ws.column_dimensions[get_column_letter(col)].width = 20 def create_master_excel(): wb = openpyxl.Workbook() wb.remove(wb.active) for sheet_name, cfg in _MASTER_SHEETS.items(): ws = wb.create_sheet(sheet_name) _write_headers(ws, cfg["headers"]) for row in cfg["rows"]: ws.append(row) wb.save(MASTER_FILE) print(f" Created {MASTER_FILE}") def create_inputs_excel(): wb = openpyxl.Workbook() ws = wb.active ws.title = "Test_Cases" _write_headers(ws, _INPUT_HEADERS) for tc in _TEST_CASES_SEED: ws.append([tc.get(h) for h in _INPUT_HEADERS]) ws.freeze_panes = "E2" wb.save(INPUT_FILE) print(f" Created {INPUT_FILE}") # ══════════════════════════════════════════════════════════════════════════════ # SECTION 3 — EXCEL READERS # ══════════════════════════════════════════════════════════════════════════════ def _sheet_to_dicts(ws) -> list[dict]: rows = list(ws.values) if len(rows) < 2: return [] headers = [str(h) if h is not None else f"col_{i}" for i, h in enumerate(rows[0])] out = [] for row in rows[1:]: if all(v is None for v in row): continue out.append(dict(zip(headers, row))) return out def load_master() -> dict: wb = openpyxl.load_workbook(MASTER_FILE, read_only=True, data_only=True) master: dict = {} for name in wb.sheetnames: rows = _sheet_to_dicts(wb[name]) if name == "Currency": master["currency_seed_map"] = { r["currency_code"]: {"minor_unit_value": str(r["minor_unit_value"])} for r in rows if r.get("currency_code") } elif name == "Entity": master["entity"] = SimpleNamespace(**rows[0]) if rows else SimpleNamespace() else: master[name.lower()] = [SimpleNamespace(**r) for r in rows] wb.close() return master def load_test_cases() -> list[dict]: wb = openpyxl.load_workbook(INPUT_FILE, read_only=True, data_only=True) rows = _sheet_to_dicts(wb.active) wb.close() return rows # ══════════════════════════════════════════════════════════════════════════════ # SECTION 4 — TYPE HELPERS # ══════════════════════════════════════════════════════════════════════════════ def _d(v) -> Decimal | None: if v is None or str(v).strip() == "": return None return Decimal(str(v)) def _dt(v) -> date | None: if v is None or str(v).strip() == "": return None if isinstance(v, datetime): return v.date() if isinstance(v, date): return v return date.fromisoformat(str(v)[:10]) def _b(v) -> bool: if isinstance(v, bool): return v if isinstance(v, (int, float)): return bool(v) return str(v).strip().upper() in {"TRUE", "YES", "1"} def _i(v) -> int | None: if v is None or str(v).strip() == "": return None return int(float(str(v))) def _s(v) -> str | None: if v is None or str(v).strip() == "": return None return str(v).strip() def _canon(s: str | None) -> str | None: if not s: return None s = unicodedata.normalize("NFKC", str(s)).upper() s = re.sub(r"[^A-Z0-9]+", "", s) return s or None # ══════════════════════════════════════════════════════════════════════════════ # SECTION 5 — OBJECT BUILDERS # ══════════════════════════════════════════════════════════════════════════════ def build_document(row: dict): return SimpleNamespace( id=_i(row.get("doc_id")), tenant_id=_i(row.get("tenant_id")), entity_id=_i(row.get("entity_id")), status=_s(row.get("doc_status")) or "validated", doc_type=_s(row.get("doc_type")) or "invoice", tenant_timezone=_s(row.get("tenant_timezone")) or "UTC", ) def build_extracted_fields(row: dict): return SimpleNamespace( sender_name=_s(row.get("sender_name")), sender_name_canonical=_s(row.get("sender_name_canonical")), sender_tax_id=_s(row.get("sender_tax_id")), sender_tax_id_canonical=_s(row.get("sender_tax_id_canonical")), bank_iban=_s(row.get("bank_iban")), bank_iban_canonical=_s(row.get("bank_iban_canonical")), bank_acc_no=_s(row.get("bank_acc_no")), bank_swift=_s(row.get("bank_swift")), bank_acc_name=_s(row.get("bank_acc_name")), invoice_date=_dt(row.get("invoice_date")), due_date=_dt(row.get("due_date")), due_date_origin=_s(row.get("due_date_origin")), service_start_date=_dt(row.get("service_start_date")), service_end_date=_dt(row.get("service_end_date")), payment_terms=_i(row.get("payment_terms")), po_number_canonical=_s(row.get("po_number_canonical")), currency=_s(row.get("currency")), invoice_no=_s(row.get("invoice_no")), total_amount=_d(row.get("total_amount")), subtotal=_d(row.get("subtotal")), tax_amount=_d(row.get("tax_amount")), tax_rate=_d(row.get("tax_rate")), ) def build_validation_line_items(row: dict) -> list: return [ SimpleNamespace( line_number=1, amount=_d(row.get("line1_amount")), tax_rate_per_item=_d(row.get("line1_tax_rate")), tax_amount_per_item=_d(row.get("line1_tax_amount")), discount_amount_per_item=None, ) ] def build_validation_pos(row: dict) -> list: val_po = _s(row.get("val_po_number")) if val_po is None: return [] return [ SimpleNamespace( id=200, tenant_id=_i(row.get("tenant_id")), po_number_canonical=val_po, payment_terms=_i(row.get("val_po_payment_terms")), deleted_at=None, ) ] def build_matching_invoice(row: dict) -> InvoiceHeader: is_acme = _s(row.get("sender_name_canonical")) == "ACMECORPORATIONLTD" return InvoiceHeader( id=_i(row.get("doc_id")) or 1001, status=_s(row.get("doc_status")) or "validated", vendor_id=100 if is_acme else None, currency=_s(row.get("currency")), total_amount=_d(row.get("subtotal")), doc_type=_s(row.get("doc_type")) or "invoice", tenant_id=_i(row.get("tenant_id")), invoice_date=_dt(row.get("invoice_date")), ) def build_matching_lines(row: dict) -> list[InvoiceLine]: desc = _s(row.get("line1_description")) sku = _s(row.get("line1_sku")) return [ InvoiceLine( id=1, description=desc, description_canonical=_canon(desc), sku=sku, sku_canonical=_canon(sku), quantity=_d(row.get("line1_quantity")), unit_price=_d(row.get("line1_unit_price")), amount=_d(row.get("line1_amount")), ) ] def build_matching_po(row: dict) -> tuple: if not _b(row.get("has_po")): return None, None po_status = _s(row.get("po_status")) or "open" po_line_status = _s(row.get("po_line_status")) or "open" po = POHeader(id=200, vendor_id=100, currency="GBP", status=po_status) lines = [ POLine( id=201, description="Software License - Annual", description_canonical="SOFTWARELICENSEANNUAL", sku="SW-LIC-001", sku_canonical="SWLIC001", quantity=Decimal("10"), unit_price=Decimal("10.00"), item_type="services", status=po_line_status, ) ] return po, lines def build_matching_gr(row: dict) -> list[GRLineItem]: if not _b(row.get("has_gr")): return [] return [ GRLineItem( id=300, po_line_id=201, gr_date=date(2025, 9, 28), inspection_status="accepted", quantity_accepted=Decimal("10"), quantity_received=Decimal("10"), ) ] def build_contract_objects(row: dict) -> tuple: """Returns (contracts_list, contract_invoice | None).""" if not _b(row.get("use_contract")): return [], None rate_card = [ RateCardItem( id=401, description="Software License - Annual", description_canonical="SOFTWARELICENSEANNUAL", unit_price=Decimal("10.00"), item_type="services", ) ] contracts = [ ContractRecord( id=400, tenant_id=_i(row.get("tenant_id")) or 1, vendor_id=100, entity_id=_i(row.get("entity_id")) or 10, currency=_s(row.get("currency")) or "GBP", status="active", effective_from=date(2024, 1, 1), rate_card_items=rate_card, ) ] contract_invoice = ContractInvoiceHeader( id=_i(row.get("doc_id")) or 1001, tenant_id=_i(row.get("tenant_id")) or 1, entity_id=_i(row.get("entity_id")) or 10, resolved_vendor_id=100, currency=_s(row.get("currency")), invoice_date=_dt(row.get("invoice_date")), service_start_date=_dt(row.get("service_start_date")), service_end_date=_dt(row.get("service_end_date")), total_amount=_d(row.get("total_amount")), ) return contracts, contract_invoice # ══════════════════════════════════════════════════════════════════════════════ # SECTION 6 — EXPECTED CHECKS # ══════════════════════════════════════════════════════════════════════════════ def _flag_codes(flags: list) -> list[str]: codes = [] for f in flags or []: if isinstance(f, dict): codes.append(str(f.get("code", ""))) else: codes.append(str(f)) return codes def check_expected(result: dict, row: dict) -> tuple[bool, list[str]]: failures: list[str] = [] codes = _flag_codes(result.get("risk_flags", [])) raw_vid = row.get("exp_resolved_vendor_id") if raw_vid is not None and str(raw_vid).strip() not in ("", "None"): try: exp_vid = int(float(str(raw_vid))) got_vid = result.get("resolved_vendor_id") if got_vid != exp_vid: failures.append( f"resolved_vendor_id: expected {exp_vid}, got {got_vid}" ) except (ValueError, TypeError): pass exp_route = _s(row.get("exp_route")) if exp_route and result.get("route") != exp_route: failures.append(f"route: expected {exp_route!r}, got {result.get('route')!r}") exp_present = _s(row.get("exp_risk_flags_present")) if exp_present: for code in [c.strip() for c in exp_present.split(",") if c.strip()]: if code not in codes: failures.append(f"expected risk flag {code!r} not found (got {codes})") exp_absent = _s(row.get("exp_risk_flags_absent")) if exp_absent: for code in [c.strip() for c in exp_absent.split(",") if c.strip()]: if code in codes: failures.append(f"risk flag {code!r} should be absent (found in {codes})") exp_exc = _s(row.get("exp_match_exception_type")) if exp_exc: mr = result.get("match_result") header_excs = list(getattr(mr, "header_exceptions", None) or []) exc_types = [e.exception_type for e in header_excs] if exp_exc not in exc_types: failures.append( f"expected match exception {exp_exc!r}, found {exc_types}" ) return len(failures) == 0, failures # ══════════════════════════════════════════════════════════════════════════════ # SECTION 7 — TEST RUNNER # ══════════════════════════════════════════════════════════════════════════════ def run_test(row: dict, master: dict) -> dict: tc_id = str(row.get("tc_id", "?")) result_row: dict = { "tc_id": tc_id, "scenario": row.get("scenario", ""), "passed": False, "failures": "", "route": "", "final_status": "", "resolved_vendor_id": "", "risk_flags": "", "advisory_flags": "", "match_overall_status": "", "match_zone": "", "header_exceptions": "", "error": "", } try: contracts, contract_invoice = build_contract_objects(row) po, po_lines = build_matching_po(row) gr_rows = build_matching_gr(row) result = run_invoice_pipeline( document=build_document(row), extracted_fields=build_extracted_fields(row), vendors=master.get("vendors", []), vendor_bank_accounts=master.get("vendor_bank_accounts", []), vendor_tax_ids=master.get("vendor_tax_ids", []), entity=master.get("entity", SimpleNamespace()), line_items=build_validation_line_items(row), purchase_orders=build_validation_pos(row), contracts=contracts, tax_master_rows=master.get("tax_master", []), currency_seed_map=master.get("currency_seed_map", {}), matching_invoice=build_matching_invoice(row), matching_inv_lines=build_matching_lines(row), contract_invoice=contract_invoice, po=po, po_lines=po_lines, gr_rows=gr_rows, ) passed, failures = check_expected(result, row) mr = result.get("match_result") header_excs = list(getattr(mr, "header_exceptions", None) or []) result_row.update({ "passed": passed, "failures": "; ".join(failures), "route": result.get("route", ""), "final_status": result.get("final_status", ""), "resolved_vendor_id": str(result.get("resolved_vendor_id", "")), "risk_flags": ", ".join(_flag_codes(result.get("risk_flags", []))), "advisory_flags": ", ".join(_flag_codes(result.get("advisory_flags", []))), "match_overall_status": getattr(mr, "overall_status", ""), "match_zone": getattr(mr, "match_zone", ""), "header_exceptions": "; ".join(e.exception_type for e in header_excs), }) tag = "PASS" if passed else "FAIL" print(f" [{tag}] {tc_id}: {row.get('scenario', '')}") for f in failures: print(f" x {f}") except Exception: result_row["error"] = traceback.format_exc(limit=6) print(f" [ERR ] {tc_id}: {result_row['error'].splitlines()[-1]}") return result_row # ══════════════════════════════════════════════════════════════════════════════ # SECTION 8 — RESULTS WRITER # ══════════════════════════════════════════════════════════════════════════════ _RES_HEADERS = [ "tc_id", "scenario", "passed", "failures", "route", "final_status", "resolved_vendor_id", "risk_flags", "advisory_flags", "match_overall_status", "match_zone", "header_exceptions", "error", ] _RES_WIDTHS = [10, 42, 8, 55, 12, 14, 20, 65, 45, 20, 12, 45, 80] _GREEN = PatternFill("solid", fgColor="C6EFCE") _RED = PatternFill("solid", fgColor="FFC7CE") _AMBER = PatternFill("solid", fgColor="FFEB9C") _BLUE = PatternFill("solid", fgColor="1F4E79") def write_results(result_rows: list[dict]): wb = openpyxl.Workbook() ws = wb.active ws.title = "Results" ws.append(_RES_HEADERS) for col in range(1, len(_RES_HEADERS) + 1): cell = ws.cell(row=1, column=col) cell.fill = _BLUE cell.font = Font(bold=True, color="FFFFFF", size=10) cell.alignment = Alignment(horizontal="center") total, n_pass = 0, 0 for rr in result_rows: ws.append([rr.get(h, "") for h in _RES_HEADERS]) r = ws.max_row total += 1 ok = bool(rr.get("passed")) if ok: n_pass += 1 fill = _GREEN if ok else (_AMBER if rr.get("error") else _RED) for col in (1, 2, 3): ws.cell(r, col).fill = fill for col, w in enumerate(_RES_WIDTHS, 1): ws.column_dimensions[get_column_letter(col)].width = w ws.freeze_panes = "A2" ws.append([]) ws.append(["", f"PASSED {n_pass} / {total}"]) sr = ws.max_row ws.cell(sr, 2).font = Font(bold=True, size=12) ws.cell(sr, 2).fill = _GREEN if n_pass == total else _RED wb.save(RESULTS_FILE) print(f"\n Results -> {RESULTS_FILE}") print(f" Summary : {n_pass}/{total} passed") # ══════════════════════════════════════════════════════════════════════════════ # SECTION 9 — MAIN # ══════════════════════════════════════════════════════════════════════════════ def main(): print("Invoice Pipeline Tests") print("-" * 60) if not os.path.exists(MASTER_FILE): print("Creating master_data.xlsx ...") create_master_excel() if not os.path.exists(INPUT_FILE): print("Creating invoice_inputs.xlsx ...") create_inputs_excel() print("Loading master data ...") master = load_master() print("Loading test cases ...") test_cases = load_test_cases() print(f" {len(test_cases)} test cases\n") print("Running tests ...") result_rows = [run_test(row, master) for row in test_cases] print("\nWriting results ...") write_results(result_rows) if __name__ == "__main__": main()