"""Contract checks for synchronous submit_v1.""" from __future__ import annotations import base64 import json import sys import unittest from pathlib import Path from unittest.mock import call, patch PROJECT_ROOT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(PROJECT_ROOT)) loaded_platform = sys.modules.get("platform") if loaded_platform is not None and not hasattr(loaded_platform, "__path__"): del sys.modules["platform"] import apps.wp_invoices as app from apps.wp_invoices.services.submission import ( confirm_invoice_supplier_profile, read_invoice_result, submit_invoice, submit_invoice_v2, ) from bridge_platform.storage.service import StorageQuotaExceeded, StorageSnapshot from bridge_platform.quotas.service import QuotaDecision, QuotaExceeded from datetime import datetime, timedelta PDF = b"%PDF-1.4\nfixture" BASE_CONTEXT = { "request_id": "request-1", "tenant": {"tenant_id": "tenant-a"}, "payload": { "filename": "invoice.pdf", "content_type": "application/pdf", "file_bytes_base64": base64.b64encode(PDF).decode("ascii"), "idempotency_key": "wp-test-0001", }, } CANONICAL = { "invoice_id": "placeholder", "status": "completed", "document": {}, "supplier": {}, "invoice_number": None, "invoice_date": None, "currency": "AUD", "items": [], "subtotal": None, "tax": None, "total": None, "validations": {}, "warnings": [], "confidence": {}, "processing": {}, } class SubmissionChecks(unittest.TestCase): def test_manifest_exports_submit_contract(self): self.assertEqual([], app.APP_MANIFEST.validate()) contract = app.APP_MANIFEST.actions["submit_v1"] self.assertEqual("invoice.document.submit", contract.capability) self.assertTrue(contract.side_effects) self.assertTrue(contract.idempotency_key_required) self.assertEqual("restricted", contract.data_classification) self.assertEqual("invoice.document.submit_trusted", app.APP_MANIFEST.actions["submit_trusted_v1"].capability) @patch("apps.wp_invoices.services.submission.write_bytes") @patch("apps.wp_invoices.services.submission.extract_canonical_invoice") def test_submission_uses_tenant_storage_and_returns_artifact_ids(self, extract, write): extract.side_effect = lambda **kwargs: {**CANONICAL, "invoice_id": kwargs["invoice_id"]} data = submit_invoice(BASE_CONTEXT) invoice_id = data["invoice_id"] self.assertTrue(invoice_id.startswith("inv_")) self.assertEqual("completed", data["status"]) self.assertNotIn("/var/", json.dumps(data)) self.assertEqual(f"invoice:{invoice_id}:original", data["artifacts"]["original"]) self.assertEqual(2, write.call_count) first = write.call_args_list[0].kwargs second = write.call_args_list[1].kwargs self.assertEqual("tenant-a", first["tenant_id"]) self.assertEqual(Path("documents") / invoice_id / "original.pdf", first["relative_path"]) self.assertEqual(PDF, first["data"]) self.assertEqual(Path("results") / invoice_id / "result.json", second["relative_path"]) extract.assert_called_once_with( file_bytes=PDF, filename="invoice.pdf", content_type="application/pdf", invoice_id=invoice_id, artifact_id=f"invoice:{invoice_id}:original", engine="mini", context=BASE_CONTEXT, ) @patch("apps.wp_invoices.services.submission.write_bytes") @patch("apps.wp_invoices.services.submission.extract_canonical_invoice") def test_same_tenant_and_idempotency_key_produce_same_invoice_id(self, extract, write): extract.side_effect = lambda **kwargs: {**CANONICAL, "invoice_id": kwargs["invoice_id"]} first = submit_invoice(BASE_CONTEXT) second = submit_invoice(BASE_CONTEXT) self.assertEqual(first["invoice_id"], second["invoice_id"]) def test_mime_signature_mismatch_is_rejected_before_storage(self): context = { **BASE_CONTEXT, "payload": { **BASE_CONTEXT["payload"], "file_bytes_base64": base64.b64encode(b"not-a-pdf").decode("ascii"), }, } with patch("apps.wp_invoices.services.submission.write_bytes") as write: with self.assertRaisesRegex(ValueError, "unsupported_document_type"): submit_invoice(context) write.assert_not_called() def test_submit_handler_returns_stable_invalid_input_error(self): context = {**BASE_CONTEXT, "payload": {}} payload, status = app.handle_request(context, "submit_v1") self.assertEqual(400, status) self.assertEqual("invalid_input", payload["error"]["code"]) self.assertNotIn("traceback", json.dumps(payload).lower()) def test_submit_handler_redacts_storage_details(self): snapshot = StorageSnapshot( tenant_id="tenant-a", storage_key="tenant-a", root="/private/root", used_bytes=10, limit_bytes=10, remaining_bytes=0, percentage=100.0, status="blocked", ) with patch( "apps.wp_invoices.services.submission.submit_invoice", side_effect=StorageQuotaExceeded(snapshot, len(PDF)), ): payload, status = app.handle_request(BASE_CONTEXT, "submit_v1") self.assertEqual(413, status) self.assertEqual("storage_quota_exceeded", payload["error"]["code"]) self.assertNotIn("/private/root", json.dumps(payload)) self.assertNotIn("tenant-a", json.dumps(payload)) def test_submit_handler_redacts_extraction_exception(self): with patch( "apps.wp_invoices.services.submission.submit_invoice", side_effect=RuntimeError("provider secret and /private/path"), ): payload, status = app.handle_request(BASE_CONTEXT, "submit_v1") self.assertEqual(502, status) self.assertEqual("extraction_failed", payload["error"]["code"]) self.assertNotIn("provider secret", json.dumps(payload)) self.assertNotIn("/private/path", json.dumps(payload)) @patch("apps.wp_invoices.services.submission.stage_candidate") @patch("apps.wp_invoices.services.submission.write_bytes") @patch("apps.wp_invoices.services.submission.extract_canonical_invoice_v2") @patch("apps.wp_invoices.services.submission._cached_trusted_result", return_value=None) def test_trusted_submission_uses_v2_and_exposes_safe_profile_state(self, cached, extract, write, stage): trusted = {**CANONICAL, "result_contract_version": "2.0", "review_status": "accepted", "quality": {}} extract.side_effect = lambda **kwargs: {**trusted, "invoice_id": kwargs["invoice_id"]} stage.return_value = {"profile_id": "sp_" + "a" * 32, "display_name": "Supplier", "status": "candidate", "version": 1, "invoice_type": "corporate", "supplier_key": "a" * 64} data = submit_invoice_v2(BASE_CONTEXT) self.assertEqual("2.0", data["result"]["result_contract_version"]) self.assertEqual("candidate", data["supplier_profile"]["status"]) self.assertNotIn("supplier_key", data["supplier_profile"]) extract.assert_called_once() @patch("apps.wp_invoices.services.submission.stage_candidate", return_value=None) @patch("apps.wp_invoices.services.submission._cached_trusted_result") def test_trusted_submission_reuses_current_extractor_result_without_ai_or_write(self, cached, stage): cached.return_value = {**CANONICAL, "processing": {"extractor_version": "unified-v2.3-ocr-layout", "canonical_revision": "monetary-column-v1"}} with patch("apps.wp_invoices.services.submission.extract_canonical_invoice_v2") as extract, \ patch("apps.wp_invoices.services.submission.write_bytes") as write: data = submit_invoice_v2(BASE_CONTEXT) self.assertEqual("unified-v2.3-ocr-layout", data["result"]["processing"]["extractor_version"]) extract.assert_not_called(); write.assert_not_called() @patch("apps.wp_invoices.services.submission.read_bytes") def test_result_read_is_tenant_scoped(self, read): read.return_value = json.dumps({**CANONICAL, "invoice_id": "inv_" + "a" * 32}).encode() context = {**BASE_CONTEXT, "payload": {"invoice_id": "inv_" + "a" * 32}} data = read_invoice_result(context) self.assertEqual("inv_" + "a" * 32, data["invoice_id"]) self.assertEqual("tenant-a", read.call_args.kwargs["tenant_id"]) def test_profile_confirmation_requires_authenticated_identity(self): context = {**BASE_CONTEXT, "payload": {"invoice_id": "inv_" + "a" * 32, "corrections": {}}} with self.assertRaises(PermissionError): confirm_invoice_supplier_profile(context) def test_trusted_handler_returns_specific_ai_quota_error(self): now = datetime.utcnow() decision = QuotaDecision( metric="ai_tokens", limit_name="ai_tokens_monthly", limit=100000, used=80544, remaining=19456, period_start=now, period_end=now + timedelta(days=1), allowed=False, policy="block", ) with patch("apps.wp_invoices.services.submission.submit_invoice_v2", side_effect=QuotaExceeded(decision)): payload, status = app.handle_request(BASE_CONTEXT, "submit_trusted_v1") self.assertEqual(429, status) self.assertEqual("ai_quota_exceeded", payload["error"]["code"]) self.assertNotIn("80544", json.dumps(payload)) if __name__ == "__main__": unittest.main(verbosity=2)