import asyncio from pathlib import Path from typing import Any import pytest from a2a_pack import LocalRunContext, PlatformUserAuth, UploadedFile from a2a_pack.cli.local import load_local_project from a2a_pack.workspace import LocalWorkspaceClient import agent as quote_agent from agent import QuoteJudgeStudioV1 ROOT = Path(__file__).resolve().parents[1] SUCCESS_ARGS = { "comparison_id": "studio-quote-v1", "quotes": [ { "delivery_days": 12, "quantity": 500, "unit_price": 10.75, "vendor": "Acme Bearings", "warranty_months": 12, }, { "delivery_days": 8, "quantity": 500, "unit_price": 9.1, "vendor": "Beta Industrial", "warranty_months": 18, }, ], "weights": {"delivery": 30, "price": 50, "warranty": 20}, } def ctx() -> LocalRunContext[PlatformUserAuth]: return LocalRunContext( auth=PlatformUserAuth(sub="user-123", user_id=123, email="buyer@example.com"), task_id="test-task", ) 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") browser_client = (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 "QuoteJudge Studio" in frontend assert "compare_quotes_browser_upload" in frontend assert "data_base64" in browser_client assert "DATABASE_URL" not in frontend + browser_client assert ".svc.cluster.local" not in frontend + browser_client 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() assert card.version == "0.1.0" skill_names = {skill.name for skill in card.skills} assert {"compare_quotes", "get_comparison", "compare_quotes_file", "compare_quotes_browser_upload"} <= skill_names databases = card.runtime.platform_resources.databases assert databases, "live Agent Card must declare its managed database" assert databases[0].name == "quote-judge-studio-v1-data" assert databases[0].scope == "user" assert databases[0].migrations is not None assert databases[0].migrations.path == "db/migrations" assert card.runtime.pricing.caller_pays_llm is False compare_schema = next(skill.input_schema for skill in card.skills if skill.name == "compare_quotes") assert set(compare_schema["required"]) == {"comparison_id", "quotes", "weights"} assert compare_schema["properties"]["quotes"]["type"] == "array" upload_schema = next(skill.input_schema for skill in card.skills if skill.name == "compare_quotes_file") assert upload_schema["properties"]["document"]["x-a2a-file-upload"]["max_bytes"] == quote_agent.MAX_UPLOAD_BYTES @pytest.mark.asyncio async def test_success_failure_reload_and_receipt(monkeypatch): saved: dict[tuple[str, str], dict[str, Any]] = {} receipts: list[dict[str, Any]] = [] async def fake_save(tenant: str, result: dict[str, Any]) -> None: saved[(tenant, result["comparison_id"])] = result async def fake_load(tenant: str, comparison_id: str) -> dict[str, Any] | None: return saved.get((tenant, comparison_id)) async def fake_receipt(ctx, skill_name, comparison_id, status, payload): receipts.append({"skill_name": skill_name, "comparison_id": comparison_id, "status": status, "payload": payload}) monkeypatch.setattr(quote_agent, "_save_comparison", fake_save) monkeypatch.setattr(quote_agent, "_load_comparison", fake_load) monkeypatch.setattr(quote_agent, "_persist_receipt_best_effort", fake_receipt) app = QuoteJudgeStudioV1() success = await app.local_invoke("compare_quotes", auth=ctx().auth, **SUCCESS_ARGS) assert success["ok"] is True assert success["comparison_id"] == "studio-quote-v1" assert success["recommendation"]["vendor"] == "Beta Industrial" assert success["execution_receipt"]["persisted"] is True reload = await app.local_invoke("get_comparison", auth=ctx().auth, comparison_id="studio-quote-v1") assert reload["ok"] is True assert reload["recommendation"]["vendor"] == "Beta Industrial" failure_args = { "comparison_id": "studio-quote-invalid", "quotes": [ { "delivery_days": 5, "quantity": 1, "unit_price": 10, "vendor": "Only Vendor", "warranty_months": 6, } ], "weights": {"delivery": 30, "price": 50, "warranty": 20}, } failure = await app.local_invoke("compare_quotes", auth=ctx().auth, **failure_args) assert failure["ok"] is False assert failure["code"] == "at_least_two_quotes_required" assert any(r["skill_name"] == "compare_quotes" and r["status"] == "ok" for r in receipts) @pytest.mark.asyncio async def test_browser_upload_bridge(monkeypatch): async def fake_save(tenant: str, result: dict[str, Any]) -> None: return None async def fake_receipt(ctx, skill_name, comparison_id, status, payload): return None monkeypatch.setattr(quote_agent, "_save_comparison", fake_save) monkeypatch.setattr(quote_agent, "_persist_receipt_best_effort", fake_receipt) csv_text = "vendor,unit_price,quantity,delivery_days,warranty_months\nAcme Bearings,10.75,500,12,12\nBeta Industrial,9.1,500,8,18\n" import base64 app = QuoteJudgeStudioV1() result = await app.local_invoke( "compare_quotes_browser_upload", auth=ctx().auth, comparison_id="studio-quote-v1", weights={"delivery": 30, "price": 50, "warranty": 20}, documents=[ { "filename": "quotes.csv", "media_type": "text/csv", "data_base64": base64.b64encode(csv_text.encode()).decode(), } ], ) assert result["ok"] is True assert result["recommendation"]["vendor"] == "Beta Industrial" bad = await app.local_invoke( "compare_quotes_browser_upload", auth=ctx().auth, comparison_id="studio-quote-v1", weights={"delivery": 30, "price": 50, "warranty": 20}, documents=[{"filename": "bad.csv", "media_type": "text/csv", "data_base64": "not base64"}], ) assert bad["ok"] is False assert bad["code"] == "invalid_base64_upload" @pytest.mark.asyncio async def test_external_file_upload_path(monkeypatch): async def fake_save(tenant: str, result: dict[str, Any]) -> None: return None async def fake_receipt(ctx, skill_name, comparison_id, status, payload): return None monkeypatch.setattr(quote_agent, "_save_comparison", fake_save) monkeypatch.setattr(quote_agent, "_persist_receipt_best_effort", fake_receipt) workspace = LocalWorkspaceClient( { "uploads/quotes.csv": b"vendor,unit_price,quantity,delivery_days,warranty_months\nAcme Bearings,10.75,500,12,12\nBeta Industrial,9.1,500,8,18\n" }, access=QuoteJudgeStudioV1.workspace_access, ) local_ctx = LocalRunContext( auth=ctx().auth, workspace=workspace, task_id="file-test", ) result = await QuoteJudgeStudioV1().invoke( "compare_quotes_file", local_ctx, comparison_id="studio-quote-v1", weights={"delivery": 30, "price": 50, "warranty": 20}, document=UploadedFile(path="uploads/quotes.csv", filename="quotes.csv", media_type="text/csv", size_bytes=128), ) assert result["ok"] is True assert result["recommendation"]["vendor"] == "Beta Industrial"