From 3d3a2f15ecc0fa6e80e3d3980317d932b9f94894 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 03:14:46 +0000 Subject: [PATCH] a2a-source-edit: write tests/test_full_stack_contract.py --- tests/test_full_stack_contract.py | 143 +++++++++++++++++++++++++++++- 1 file changed, 142 insertions(+), 1 deletion(-) diff --git a/tests/test_full_stack_contract.py b/tests/test_full_stack_contract.py index 59cee34..173f390 100644 --- a/tests/test_full_stack_contract.py +++ b/tests/test_full_stack_contract.py @@ -1,28 +1,169 @@ +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 "Skill runner" not in frontend, "replace the generic scaffold with the product workflow" + 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)