"""Checks for the single-call schema-bound v2 extractor.""" from __future__ import annotations import json import sys import unittest from pathlib import Path from unittest.mock import patch from jsonschema import validate PROJECT_ROOT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(PROJECT_ROOT)) from apps.wp_invoices.services.extractor import run_unified_extraction_v2 from apps.wp_invoices.services.pipeline import extract_canonical_invoice_v2 from bridge_platform.ai.openai_client import _build_request_kwargs APP_ROOT = Path(__file__).resolve().parents[1] SCHEMA = json.loads((APP_ROOT / "contracts" / "extraction.unified.v2.schema.json").read_text()) RECOVERY_SCHEMA = json.loads((APP_ROOT / "contracts" / "items-recovery.v2.schema.json").read_text()) def assert_strict_objects(testcase, node): if isinstance(node, dict): if node.get("type") == "object" and node.get("additionalProperties") is False: testcase.assertEqual(set((node.get("properties") or {}).keys()), set(node.get("required") or [])) for value in node.values(): assert_strict_objects(testcase, value) elif isinstance(node, list): for value in node: assert_strict_objects(testcase, value) def field(value=None, confidence=0.9): return {"verbatim": value, "computed": None, "confidence": confidence} EXTRACTED = { "invoice_type": field("POS"), "document_subtype": field("TAX_INVOICE"), "observed_item_line_count": 1, "header": { "supplier": {"name": field("ALDI STORES"), "legal_entity": field(None), "abn": field(None)}, "invoice": {"number": field(None), "date": {"issue_date": field("06 APR 2024"), "due_date": field(None)}}, "store": field("SPRINGVALE SOUTH"), "transaction_time": field("16:28"), "payment_method": field("Debit Mastercard"), "payment_reference": field("908195"), }, "items": [{ "description": field("Item"), "qty": field("1"), "transaction_unit": field("each"), "package_size": {"value": None, "unit": None}, "unit_price": field("3.49"), "unit_price_basis": field("each"), "gst_line": field(None), "line_total": field("3.49"), "tax_code": field("A"), "source_text": "Item 3.49 A", "source_line": "item-1", "bounding_box": None, }], "adjustments": [{"type": "payment_surcharge", "description": "Card surcharge", "amount": 0.44, "source_text": "Total surcharge 0.44", "confidence": 0.9}], "tax_summary": {"tax_inclusive": True, "tax_total": field("1.98"), "lines": [{"tax_code": "B", "rate": 10, "net_amount": 19.8, "tax_amount": 1.98}]}, "totals": {"subtotal": field("89.46"), "gst": field(None), "grand_total": field("89.90")}, "currency": field("AUD"), "reference_candidates": [{"label": "Receipt", "value": "123", "confidence": 0.8}], } class UnifiedExtractorChecks(unittest.TestCase): def test_fixture_conforms_to_strict_extraction_schema(self): validate(instance=EXTRACTED, schema=SCHEMA) def test_recovery_schema_meets_recursive_strict_output_requirements(self): assert_strict_objects(self, RECOVERY_SCHEMA) @patch("apps.wp_invoices.services.extractor.run_llm") def test_classification_and_extraction_use_exactly_one_ai_call(self, run_llm): run_llm.return_value = {"output": EXTRACTED, "usage": {"total_tokens": 100}} response = run_unified_extraction_v2( data=b"image", filename="receipt.jpg", mimetype="image/jpeg", context={"tenant": {"tenant_id": "tenant-a"}}, ) self.assertEqual("POS", response["extracted"]["invoice_type"]["verbatim"]) run_llm.assert_called_once() options = run_llm.call_args.kwargs["options"] self.assertEqual("json_schema", options["response_format"]) self.assertTrue(options["json_schema"]["additionalProperties"] is False) @patch("apps.wp_invoices.services.extractor.run_llm") def test_invalid_structured_response_is_rejected(self, run_llm): run_llm.return_value = {"output": {"_raw": "truncated"}, "usage": {}} with self.assertRaisesRegex(ValueError, "structured_extraction_invalid"): run_unified_extraction_v2(data=b"image", filename="x.jpg", mimetype="image/jpeg") @patch("apps.wp_invoices.services.pipeline._run_isolated_monetary_recovery") @patch("apps.wp_invoices.services.pipeline.run_unified_extraction_v2") def test_v2_pipeline_recovers_low_coverage_items(self, extract, recover): first_pass = json.loads(json.dumps(EXTRACTED)) first_pass["observed_item_line_count"] = 2 layout_line = {"source_line_id": "ocr-line-1", "raw_text": "57164 Item A", "product_code": "57164", "tax_code": "A", "classification": "item_primary", "complete": False, "missing_fields": ["line_total"], "monetary_candidates": [], "confidence": 0.9, "estimated_bounding_region": [0, 0, 1, 0.1]} extract.return_value = {"extracted": first_pass, "usage": {"total_tokens": 100}, "ocr_layout": {"status": "success", "lines": [layout_line], "item_primary_lines": [layout_line]}} recover.return_value = {"patches": [{"operation": "fill_missing_line_total", "source_line_id": "ocr-line-1", "line_total": 89.46, "quantity": None, "unit_price": None, "selected_candidate": None, "confidence": 0.8, "evidence_text": "visual"}], "crop_candidates": {"ocr-line-1": [89.46]}, "usage": {"total_tokens": 50}} result = extract_canonical_invoice_v2( file_bytes=b"image", filename="receipt.jpg", content_type="image/jpeg", invoice_id="inv-1", artifact_id="artifact-1", ) self.assertEqual("pos", result["invoice_type"]) self.assertEqual("needs_review", result["review_status"]) self.assertEqual("unified-v2.3-ocr-layout", result["processing"]["extractor_version"]) recovery = result["processing"]["recovery"] self.assertEqual((True, True, False), (recovery["attempted"], recovery["accepted"], recovery["failed"])) self.assertEqual(150, result["processing"]["usage"]["total_tokens"]) extract.assert_called_once() recover.assert_called_once() @patch("apps.wp_invoices.services.pipeline._run_isolated_monetary_recovery") @patch("apps.wp_invoices.services.pipeline.run_unified_extraction_v2") def test_complete_first_pass_does_not_spend_recovery_tokens(self, extract, recover): complete = json.loads(json.dumps(EXTRACTED)) complete["items"][0]["qty"]["verbatim"] = "1" complete["items"][0]["unit_price"]["verbatim"] = "89.46" complete["items"][0]["line_total"]["verbatim"] = "89.46" extract.return_value = {"extracted": complete, "usage": {"total_tokens": 100}} result = extract_canonical_invoice_v2( file_bytes=b"image", filename="receipt.jpg", content_type="image/jpeg", invoice_id="inv-1", artifact_id="artifact-1", ) self.assertEqual("needs_review", result["review_status"]) recovery = result["processing"]["recovery"] self.assertEqual((False, False, False), (recovery["attempted"], recovery["accepted"], recovery["failed"])) recover.assert_not_called() @patch("apps.wp_invoices.services.pipeline._run_isolated_monetary_recovery") @patch("apps.wp_invoices.services.pipeline.run_unified_extraction_v2") def test_worse_recovery_does_not_replace_first_pass(self, extract, recover): extract.return_value = {"extracted": EXTRACTED, "usage": {"total_tokens": 100}} recover.return_value = {"patches": [], "crop_candidates": {}, "usage": {"total_tokens": 40}} result = extract_canonical_invoice_v2( file_bytes=b"image", filename="receipt.jpg", content_type="image/jpeg", invoice_id="inv-1", artifact_id="artifact-1", ) self.assertEqual(1, len(result["items"])) recovery = result["processing"]["recovery"] self.assertEqual((False, False, False), (recovery["attempted"], recovery["accepted"], recovery["failed"])) @patch("apps.wp_invoices.services.pipeline._run_isolated_monetary_recovery") @patch("apps.wp_invoices.services.pipeline.run_unified_extraction_v2") def test_failed_recovery_preserves_reviewable_first_pass(self, extract, recover): extract.return_value = {"extracted": EXTRACTED, "usage": {"total_tokens": 100}} recover.side_effect = RuntimeError("provider details must remain internal") result = extract_canonical_invoice_v2( file_bytes=b"image", filename="receipt.jpg", content_type="image/jpeg", invoice_id="inv-1", artifact_id="artifact-1", ) self.assertEqual("needs_review", result["review_status"]) self.assertEqual(1, len(result["items"])) recovery = result["processing"]["recovery"] self.assertEqual((False, False, False), (recovery["attempted"], recovery["accepted"], recovery["failed"])) self.assertNotIn("provider details", json.dumps(result)) @patch("apps.wp_invoices.services.pipeline.load_active_profile") @patch("apps.wp_invoices.services.pipeline._run_isolated_monetary_recovery") @patch("apps.wp_invoices.services.pipeline.run_unified_extraction_v2") def test_confirmed_profile_only_guides_conditional_recovery(self, extract, recover, load): extract.return_value = {"extracted": EXTRACTED, "usage": {}} recover.return_value = {"patches": [], "crop_candidates": {}, "usage": {}} load.return_value = { "status": "active", "invoice_type": "pos", "hints": {"invoice_number_labels": ["Receipt"], "reference_labels": [], "item_columns": ["description", "line_total"], "notes": []}, } result = extract_canonical_invoice_v2( file_bytes=b"image", filename="receipt.jpg", content_type="image/jpeg", invoice_id="inv-1", artifact_id="artifact-1", context={"tenant": {"tenant_id": "tenant-a"}}, ) self.assertTrue(result["processing"]["supplier_profile_applied"]) recover.assert_not_called() def test_platform_client_builds_strict_json_schema_request(self): request = _build_request_kwargs( prompt="extract", model="gpt-4o-mini", options={"response_format": "json_schema", "schema_name": "invoice", "json_schema": SCHEMA}, ) response_format = request["response_format"] self.assertEqual("json_schema", response_format["type"]) self.assertTrue(response_format["json_schema"]["strict"]) if __name__ == "__main__": unittest.main(verbosity=2)