import asyncio import json from pathlib import Path from unittest.mock import patch from a2a_pack import LocalRunContext, PlatformUserAuth from a2a_pack.cli.local import load_local_project from agent import ( DecisionInput, InvoiceGuardStudioV1, _parse_browser_documents, _review_payload, ) ROOT = Path(__file__).resolve().parents[1] SUCCESS_INVOICES = [ {"invoice_number": "INV-100", "subtotal": 90, "tax": 10, "total": 100, "vendor": "Acme"}, {"invoice_number": "INV-100", "subtotal": 90, "tax": 10, "total": 100, "vendor": "Acme"}, {"invoice_number": "INV-200", "subtotal": 50, "tax": 5, "total": 80, "vendor": "Beta"}, ] def test_full_stack_product_contract(): manifest = (ROOT / "a2a.yaml").read_text(encoding="utf-8") source = (ROOT / "agent.py").read_text(encoding="utf-8") frontend = (ROOT / "frontend" / "src" / "App.jsx").read_text(encoding="utf-8") bridge = (ROOT / "frontend" / "src" / "a2a.js").read_text(encoding="utf-8") assert "frontend:" in manifest and "mount: /app" in manifest assert f"version: {InvoiceGuardStudioV1.version}" in manifest assert "public: false" in manifest assert "resources:" in manifest and "databases:" in manifest assert "migrations:" in manifest and "db/migrations" in manifest assert "PlatformUserAuth" in source assert "AgentPlatformResources" in source and "AgentDatabase(" in source assert "LLMProvisioning" not in source and "ctx.llm" not in source assert "Skill runner" not in frontend assert "InvoiceGuard Studio" in frontend assert "fileToBrowserDocument" in bridge assert "A2A_LITELLM_KEY" not in source + frontend + bridge assert "DATABASE_URL" not in frontend + bridge migrations = list((ROOT / "db" / "migrations").glob("*.sql")) assert migrations and all(path.read_text(encoding="utf-8").strip() for path in migrations) card = load_local_project(ROOT).agent_cls().card() skills = {skill.name: skill for skill in card.skills} for name in {"review_invoices", "get_invoice_case", "record_decision", "review_invoice_uploads", "review_invoice_file"}: assert name in skills assert set(skills["review_invoices"].input_schema["required"]) == {"case_id", "invoices"} assert "document" in skills["review_invoice_file"].input_schema["properties"] assert card.runtime.pricing.caller_pays_llm is False assert card.runtime.platform_resources.databases assert card.runtime.platform_resources.databases[0].migrations.path == "db/migrations" assert card.workspace_access.enabled is True assert [mode.value for mode in card.workspace_access.allowed_modes] == ["read_only"] assert skills["record_decision"].policy.idempotent is False assert skills["review_invoices"].input_schema["properties"]["case_id"]["maxLength"] == 120 def test_review_success_fixture_is_deterministic(): result = _review_payload("studio-invoice-v1", SUCCESS_INVOICES) assert result["ok"] is True assert result["case_id"] == "studio-invoice-v1" assert result["duplicate_count"] == 1 assert result["total_mismatch_count"] == 1 assert result["duplicates"][0]["invoice_number"] == "INV-100" assert result["total_mismatches"][0]["invoice_number"] == "INV-200" assert result["total_mismatches"][0]["expected_total"] == 55.0 def test_incomplete_invoice_failure_fixture(): result = _review_payload("studio-invoice-invalid", [{"invoice_number": "INV-X", "subtotal": 10, "tax": 1}]) assert result["ok"] is False assert result["code"] == "incomplete_invoice" assert "vendor" in result["message"] and "total" in result["message"] blank = _review_payload("studio-invoice-invalid", [{"invoice_number": " ", "vendor": " ", "subtotal": 10, "tax": 1, "total": 11}]) assert blank["ok"] is False assert blank["code"] == "incomplete_invoice" assert "invoice_number" in blank["message"] and "vendor" in blank["message"] oversized = _review_payload("studio-invoice-invalid", [{"invoice_number": "I" * 121, "vendor": "Acme", "subtotal": 10, "tax": 1, "total": 11}]) assert oversized["ok"] is False assert oversized["code"] == "invalid_invoice" def test_browser_upload_bridge_bounds_and_parses_json(): payload = json.dumps({"invoices": SUCCESS_INVOICES}).encode() import base64 result = _parse_browser_documents([ {"filename": "fixture.json", "media_type": "application/json", "data_base64": base64.b64encode(payload).decode()} ]) assert result["ok"] is True assert len(result["invoices"]) == 3 bad = _parse_browser_documents([ {"filename": "fixture.json", "media_type": "application/json", "data_base64": "not-base64!!!"} ]) assert bad["ok"] is False assert bad["code"] == "invalid_base64" def test_acceptance_calls_with_mocked_persistence_reload_and_decision(): agent = InvoiceGuardStudioV1() auth = PlatformUserAuth(sub="stable-sub", user_id=123, email="user@example.test") ctx = LocalRunContext(auth=auth) store = {} def persist(tenant, case, source): receipt = {"receipt_id": "invoice_guard:studio-invoice-v1:test", "kind": "production_execution_receipt"} saved = {**case, "receipt": receipt, "saved_at": "2026-01-01T00:00:00+00:00", "updated_at": "2026-01-01T00:00:00+00:00"} store[(tenant, case["case_id"])] = saved return {"receipt": receipt, "saved_at": saved["saved_at"], "case": saved} def load(tenant, case_id): return store.get((tenant, case_id)) def decision(tenant, case_id, payload): case = store.get((tenant, case_id)) if case is None: return None case = {**case, "decisions": [{**payload, "decided_at": "2026-01-01T00:00:01+00:00"}]} store[(tenant, case_id)] = case return case with patch("agent._persist_case_with_receipt", side_effect=persist), patch("agent._load_case", side_effect=load), patch("agent._persist_decision", side_effect=decision): success = asyncio.run(agent.review_invoices(ctx, "studio-invoice-v1", SUCCESS_INVOICES)) assert success["ok"] is True assert success["duplicate_count"] == 1 assert success["total_mismatch_count"] == 1 assert success["receipt"]["kind"] == "production_execution_receipt" reload = asyncio.run(agent.get_invoice_case(ctx, "studio-invoice-v1")) assert reload["ok"] is True assert reload["case_id"] == "studio-invoice-v1" assert reload["duplicate_count"] == 1 assert reload["total_mismatch_count"] == 1 updated = asyncio.run(agent.record_decision(ctx, "studio-invoice-v1", DecisionInput(decision_id="d1", decision="hold", note="Investigate duplicate", invoice_number="INV-100"))) assert updated["ok"] is True assert updated["decisions"][0]["decision"] == "hold" invalid_reload = asyncio.run(agent.get_invoice_case(ctx, "bad case id")) assert invalid_reload["ok"] is False assert invalid_reload["code"] == "invalid_case_id" invalid_decision = asyncio.run( agent.record_decision( ctx, "bad case id", DecisionInput(decision_id="d2", decision="hold"), ) ) assert invalid_decision["ok"] is False assert invalid_decision["code"] == "invalid_case_id"