159 lines
7.1 KiB
Python
159 lines
7.1 KiB
Python
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 "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"
|