import asyncio import base64 import json from pathlib import Path from a2a_pack import PlatformUserAuth from a2a_pack.cli.local import load_local_project from a2a_pack.mcp import skills_to_tools from agent import QuoteJudgeStudioV1, _LOCAL_COMPARISONS, _LOCAL_RECEIPTS 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}, } AUTH = PlatformUserAuth(sub="test-user", user_id=42, email="buyer@example.test") def run(coro): return asyncio.run(coro) 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") a2a_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 "llm_provisioning" not in source assert "ctx.llm" not in source assert "Skill runner" not in frontend assert "QuoteJudge" in frontend and "compare_quotes" in frontend assert "config.json" in a2a_client and "authorizeUrl" in a2a_client forbidden = ["OPENAI_API_KEY", "A2A_LITELLM_KEY", "DATABASE_URL", ".svc.cluster.local"] assert not any(secret in frontend + a2a_client for secret in forbidden) 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() skill_names = {skill.name for skill in card.skills} assert {"compare_quotes", "get_comparison", "compare_uploaded_quotes_browser", "compare_uploaded_quote_file"} <= skill_names 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_uploaded_quote_file") assert upload_schema["properties"]["document"]["x-a2a-file-upload"]["max_bytes"] == 128000 databases = card.runtime.platform_resources.databases assert databases, "live Agent Card must declare its managed database" assert databases[0].scope == "user" assert databases[0].migrations is not None assert databases[0].migrations.path == "db/migrations" def test_success_failure_reload_and_receipt(): _LOCAL_COMPARISONS.clear() _LOCAL_RECEIPTS.clear() agent = QuoteJudgeStudioV1() success = run(agent.local_invoke("compare_quotes", auth=AUTH, **SUCCESS_ARGS)) assert success["ok"] is True assert success["comparison_id"] == "studio-quote-v1" assert success["recommendation"]["vendor"] == "Beta Industrial" assert success["recommendation"]["normalized_total"] == 4550.0 assert success["quotes"][0]["vendor"] == "Beta Industrial" assert success["quotes"][0]["scores"]["weighted_total"] == 1.0 reload = run(agent.local_invoke("get_comparison", auth=AUTH, comparison_id="studio-quote-v1")) assert reload["ok"] is True assert reload["comparison_id"] == "studio-quote-v1" assert reload["recommendation"]["vendor"] == "Beta Industrial" failure = run( agent.local_invoke( "compare_quotes", auth=AUTH, 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}, ) ) assert failure["ok"] is False assert failure["code"] == "at_least_two_quotes_required" assert len(_LOCAL_RECEIPTS) >= 3 assert any(r["comparison_id"] == "studio-quote-v1" and r["status"] == "ok" for r in _LOCAL_RECEIPTS) assert all("input_hash" in r and "result_hash" in r for r in _LOCAL_RECEIPTS) def test_browser_upload_bridge_success_and_bounds(): _LOCAL_COMPARISONS.clear() agent = QuoteJudgeStudioV1() csv_text = ( "vendor,quantity,unit_price,delivery_days,warranty_months\n" "Acme Bearings,500,10.75,12,12\n" "Beta Industrial,500,9.10,8,18\n" ) result = run( agent.local_invoke( "compare_uploaded_quotes_browser", auth=AUTH, comparison_id="studio-quote-upload", documents=[ { "filename": "quotes.csv", "media_type": "text/csv", "data_base64": base64.b64encode(csv_text.encode()).decode(), } ], weights={"delivery": 30, "price": 50, "warranty": 20}, ) ) assert result["ok"] is True assert result["recommendation"]["vendor"] == "Beta Industrial" invalid = run( agent.local_invoke( "compare_uploaded_quotes_browser", auth=AUTH, comparison_id="bad-upload", documents=[{"filename": "bad.csv", "media_type": "text/csv", "data_base64": "%%%"}], weights={"delivery": 30, "price": 50, "warranty": 20}, ) ) assert invalid["ok"] is False assert invalid["code"] == "invalid_base64" def test_mcp_tools_expose_contract_names(): agent = QuoteJudgeStudioV1() tools = skills_to_tools(agent.skills.values()) tool_names = {tool["name"] for tool in tools} assert "compare_quotes" in tool_names assert "get_comparison" in tool_names compare_tool = next(tool for tool in tools if tool["name"] == "compare_quotes") assert set(compare_tool["inputSchema"]["required"]) == {"comparison_id", "quotes", "weights"} # JSON-serializable MCP tool schemas are required for tools/list responses. json.dumps(tools)