""" Ad-hoc edge case tests — scenarios NOT in run_pipeline_tests.py Run: python edge_case_tests.py """ import sys, os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from datetime import date from decimal import Decimal from types import SimpleNamespace 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 # ── Shared master data ──────────────────────────────────────────────────────── VENDORS = [SimpleNamespace( id=100, tenant_id=1, legal_name="Acme Corporation Ltd", vendor_name="Acme Corporation Ltd", vendor_status="active", deleted_at=None, billing_country="GB", billing_state=None, tax_registration_number="GB123456715", payment_terms=30, exemption_status="standard", )] VENDOR_TAX_IDS = [SimpleNamespace( tenant_id=1, vendor_id=100, tax_id_canonical="123456715", tax_id_type="VAT-GB", verified_at="2024-01-01", deleted_at=None, effective_from=None, effective_to=None, )] VENDOR_BANK_ACCOUNTS = [SimpleNamespace( tenant_id=1, vendor_id=100, iban_canonical="GB29NWBK60161331926819", acc_no_canonical="31926819", swift_bic="NWBKGB2L", bank_acc_name="Acme Corporation Ltd", currency_code="GBP", bank_country="GB", is_primary=True, effective_from=None, effective_to=None, deleted_at=None, )] ENTITY = SimpleNamespace(id=1, tenant_id=1, entity_id=10, country_code="GB", vat_id="GB987654321", region_code=None) TAX_MASTER = [ SimpleNamespace(id=1, tenant_id=1, is_active=True, deleted_at=None, country_code="GB", region_code=None, tax_type="VAT", tax_rate=20.0, tax_name="VAT Standard 20%", effective_from=None, effective_to=None), SimpleNamespace(id=2, tenant_id=1, is_active=True, deleted_at=None, country_code="GB", region_code=None, tax_type="VAT", tax_rate=5.0, tax_name="VAT Reduced 5%", effective_from=None, effective_to=None), ] CURRENCY = {"GBP": {"minor_unit_value": "0.01"}, "USD": {"minor_unit_value": "0.01"}} VAL_PO = [SimpleNamespace(id=200, tenant_id=1, po_number_canonical="PO-001", payment_terms=30, deleted_at=None)] def make_doc(status="validated"): return SimpleNamespace(id=1001, tenant_id=1, entity_id=10, status=status, doc_type="invoice", tenant_timezone="Europe/London") def make_ef(**kw): base = dict( 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=date(2025, 10, 1), due_date=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-TEST", total_amount=Decimal("120.00"), subtotal=Decimal("100.00"), tax_amount=Decimal("20.00"), tax_rate=Decimal("20.0"), ) base.update(kw) return SimpleNamespace(**base) def make_val_lines(amount="100.00", tax_amount="20.00"): return [SimpleNamespace( line_number=1, amount=Decimal(amount), tax_rate_per_item=Decimal("20.0"), tax_amount_per_item=Decimal(tax_amount), discount_amount_per_item=None, )] def make_inv_header(currency="GBP", vendor_id=100, total_amount="100.00", invoice_date=date(2025, 10, 1), status="validated"): return InvoiceHeader(id=1001, status=status, vendor_id=vendor_id, currency=currency, total_amount=Decimal(total_amount), invoice_date=invoice_date) def make_inv_lines(unit_price="10.00", amount="100.00"): return [InvoiceLine(id=1, description="Software License - Annual", description_canonical="SOFTWARELICENSEANNUAL", sku="SW-LIC-001", sku_canonical="SWLIC001", quantity=Decimal("10"), unit_price=Decimal(unit_price), amount=Decimal(amount))] def make_po(status="open", line_status="open", currency="GBP", vendor_id=100, invoiced_qty="0"): ph = POHeader(id=200, vendor_id=vendor_id, currency=currency, status=status) pl = [POLine(id=201, description="Software License - Annual", description_canonical="SOFTWARELICENSEANNUAL", sku="SW-LIC-001", sku_canonical="SWLIC001", quantity=Decimal("10"), invoiced_qty=Decimal(invoiced_qty), unit_price=Decimal("10.00"), item_type="services", status=line_status)] return ph, pl def run_case(label, **kw): defaults = dict( document=make_doc(), extracted_fields=make_ef(), vendors=VENDORS, vendor_bank_accounts=VENDOR_BANK_ACCOUNTS, vendor_tax_ids=VENDOR_TAX_IDS, entity=ENTITY, line_items=make_val_lines(), purchase_orders=[], contracts=[], tax_master_rows=TAX_MASTER, currency_seed_map=CURRENCY, matching_invoice=make_inv_header(), matching_inv_lines=make_inv_lines(), contract_invoice=None, po=None, po_lines=None, gr_rows=None, ) defaults.update(kw) try: r = run_invoice_pipeline(**defaults) mr = r.get("match_result") risk = [f.get("code", "") if isinstance(f, dict) else str(f) for f in r.get("risk_flags", [])] adv = [f.get("code", "") if isinstance(f, dict) else str(f) for f in r.get("advisory_flags", [])] hexc = [e.exception_type for e in (getattr(mr, "header_exceptions", None) or [])] mexc = [e.exception_type for e in (getattr(mr, "match_exceptions", None) or [])] print(f"\n{'-'*62}") print(f" {label}") print(f" route={r.get('route')} final={r.get('final_status')} " f"zone={getattr(mr, 'match_zone', '-')} " f"match_status={getattr(mr, 'overall_status', '-')}") if risk: print(f" risk_flags : {risk}") if adv: print(f" advisory_flags: {adv}") if hexc: print(f" header_exc : {hexc}") if mexc: print(f" match_exc : {mexc}") if not risk and not adv and not hexc and not mexc: print(f" (no flags / exceptions)") except Exception as ex: print(f"\n{'-'*62}") print(f" {label}") print(f" RAISED: {type(ex).__name__}: {ex}") # ══════════════════════════════════════════════════════════════════════════════ print("=" * 62) print(" EDGE CASE TESTS") print("=" * 62) # ── Routing edge cases ──────────────────────────────────────────────────────── ph, pl = make_po(status="open", line_status="cancelled") run_case("EC01: PO open + all lines cancelled -> po_all_lines_closed", extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph, po_lines=pl) # ── Two-way header exceptions ───────────────────────────────────────────────── ph2, pl2 = make_po(currency="USD") run_case("EC02: Currency mismatch (invoice=GBP, PO=USD)", extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph2, po_lines=pl2, matching_invoice=make_inv_header(currency="GBP")) ph3, pl3 = make_po(vendor_id=999) run_case("EC03: Vendor mismatch (invoice vendor=100, PO vendor=999)", extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph3, po_lines=pl3) ph4, pl4 = make_po(invoiced_qty="10") # fully invoiced, available_qty=0 run_case("EC04: PO ceiling exhausted (invoiced_qty == quantity)", extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph4, po_lines=pl4) # ── Three-way GR checks ─────────────────────────────────────────────────────── ph5, pl5 = make_po() gr_fraud = [GRLineItem(id=300, po_line_id=201, gr_date=date(2025, 11, 1), # GR received AFTER invoice date inspection_status="accepted", quantity_accepted=Decimal("10"), quantity_received=Decimal("10"))] run_case("EC05: Timing fraud — invoice 2025-10-01 before GR 2025-11-01", extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph5, po_lines=pl5, gr_rows=gr_fraud) ph6, pl6 = make_po() gr_pending = [GRLineItem(id=300, po_line_id=201, gr_date=date(2025, 9, 28), inspection_status="pending", quantity_accepted=None, quantity_received=Decimal("10"))] run_case("EC06: No accepted GR (all pending)", extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph6, po_lines=pl6, gr_rows=gr_pending) ph7, pl7 = make_po() gr_rejected = [GRLineItem(id=300, po_line_id=201, gr_date=date(2025, 9, 28), inspection_status="rejected", quantity_accepted=Decimal("10"), quantity_received=Decimal("10"))] run_case("EC07: GR inspection rejected", extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph7, po_lines=pl7, gr_rows=gr_rejected) # ── Contract edge cases ─────────────────────────────────────────────────────── rate_card = [RateCardItem(id=401, description="Software License - Annual", description_canonical="SOFTWARELICENSEANNUAL", unit_price=Decimal("10.00"), item_type="services")] expired_c = [ContractRecord(id=400, tenant_id=1, vendor_id=100, entity_id=10, currency="GBP", status="active", effective_from=date(2020, 1, 1), effective_to=date(2021, 12, 31), rate_card_items=rate_card)] c_inv = ContractInvoiceHeader(id=1001, tenant_id=1, entity_id=10, resolved_vendor_id=100, currency="GBP", invoice_date=date(2025, 10, 1), total_amount=Decimal("120.00")) run_case("EC08: Contract expired (effective_to=2021-12-31)", contracts=expired_c, contract_invoice=c_inv) budget_c = [ContractRecord(id=400, tenant_id=1, vendor_id=100, entity_id=10, currency="GBP", status="active", effective_from=date(2024, 1, 1), total_value=Decimal("500.00"), cumulative_billed_amount=Decimal("450.00"), rate_card_items=rate_card)] c_inv2 = ContractInvoiceHeader(id=1001, tenant_id=1, entity_id=10, resolved_vendor_id=100, currency="GBP", invoice_date=date(2025, 10, 1), total_amount=Decimal("120.00")) run_case("EC09: Contract budget exceeded (450 billed + 120 invoice > 500 limit)", contracts=budget_c, contract_invoice=c_inv2) red_c = [ContractRecord(id=400, tenant_id=1, vendor_id=100, entity_id=10, currency="GBP", status="active", effective_from=date(2024, 1, 1), rate_card_items=rate_card)] c_inv3 = ContractInvoiceHeader(id=1001, tenant_id=1, entity_id=10, resolved_vendor_id=100, currency="GBP", invoice_date=date(2025, 10, 1), total_amount=Decimal("180.00")) run_case("EC10: Contract rate card red zone (invoice price 15.00 vs rate 10.00 = 50%)", contracts=red_c, contract_invoice=c_inv3, matching_invoice=make_inv_header(total_amount="150.00"), matching_inv_lines=make_inv_lines(unit_price="15.00", amount="150.00"), line_items=make_val_lines(amount="150.00", tax_amount="30.00")) # ── Vendor / auth edge cases ────────────────────────────────────────────────── deleted_v = [SimpleNamespace( id=100, tenant_id=1, legal_name="Acme Corporation Ltd", vendor_name="Acme Corporation Ltd", vendor_status="active", deleted_at="2023-01-01", billing_country="GB", billing_state=None, tax_registration_number="GB123456715", payment_terms=30, exemption_status="standard", )] ph11, pl11 = make_po() run_case("EC11: Vendor soft-deleted (deleted_at set) — should be unknown_vendor", vendors=deleted_v, extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph11, po_lines=pl11) ph12, pl12 = make_po() run_case("EC12: Invoice status=draft — should raise precondition error", document=make_doc(status="draft"), extracted_fields=make_ef(po_number_canonical="PO-001"), purchase_orders=VAL_PO, po=ph12, po_lines=pl12, matching_invoice=make_inv_header(status="draft")) print(f"\n{'='*62}") print(" Done.")