a2a-source-edit: write tests/test_full_stack_contract.py

This commit is contained in:
a2a-cloud
2026-07-18 06:06:52 +00:00
parent 23e971666e
commit 774b6500dd

View File

@@ -1,32 +1,10 @@
import asyncio
import base64
import json
from pathlib import Path
from a2a_pack import LocalRunContext, PlatformUserAuth
from a2a_pack.cli.local import load_local_project
from agent import BrowserDocument, QuoteInput, QuoteJudgeStudioV1, QuoteWeights, _parse_browser_documents
ROOT = Path(__file__).resolve().parents[1]
SUCCESS_ARGS = {
"comparison_id": "studio-quote-v1",
"quotes": [
{"vendor": "Acme Bearings", "unit_price": 10.75, "quantity": 500, "delivery_days": 12, "warranty_months": 12},
{"vendor": "Beta Industrial", "unit_price": 9.1, "quantity": 500, "delivery_days": 8, "warranty_months": 18},
],
"weights": {"price": 50, "delivery": 30, "warranty": 20},
}
def _ctx():
return LocalRunContext(
auth=PlatformUserAuth(sub="test-user", user_id=42, email="quote@example.test"),
task_id="test-task",
caller="pytest",
)
def test_full_stack_product_contract():
manifest = (ROOT / "a2a.yaml").read_text(encoding="utf-8")
@@ -34,89 +12,17 @@ def test_full_stack_product_contract():
frontend = (ROOT / "frontend" / "src" / "App.jsx").read_text(encoding="utf-8")
assert "frontend:" in manifest and "mount: /app" in manifest
assert "version: 0.1.1" 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 "A2A_LITELLM" not in frontend and "DATABASE_URL" not in frontend
assert "Skill runner" not in frontend, "replace the generic scaffold with the product workflow"
migrations = list((ROOT / "db" / "migrations").glob("*.sql"))
assert migrations and all(path.read_text(encoding="utf-8").strip() for path in migrations)
assert "005_quotejudge_receipt_compat.sql" in {path.name for path in migrations}
card = load_local_project(ROOT).agent_cls().card()
assert card.version == "0.1.1"
skill_names = {skill.name for skill in card.skills}
assert {"compare_quotes", "get_comparison", "upload_quotes", "upload_quote_file"}.issubset(skill_names)
databases = card.runtime.platform_resources.databases
assert databases, "live Agent Card must declare its managed database"
assert databases[0].migrations is not None
assert databases[0].migrations.path == "db/migrations"
assert card.runtime.pricing.caller_pays_llm is False
def test_compare_reload_failure_and_receipt():
agent = QuoteJudgeStudioV1()
ctx = _ctx()
result = asyncio.run(agent.compare_quotes(
ctx,
SUCCESS_ARGS["comparison_id"],
[QuoteInput.model_validate(item) for item in SUCCESS_ARGS["quotes"]],
QuoteWeights.model_validate(SUCCESS_ARGS["weights"]),
))
assert result["ok"] is True
assert result["comparison_id"] == "studio-quote-v1"
assert result["recommendation"]["vendor"] == "Beta Industrial"
assert result["receipt"]["persisted"] is True
assert "receipt_id" in result["receipt"]
reopened = asyncio.run(agent.get_comparison(ctx, "studio-quote-v1"))
assert reopened["ok"] is True
assert reopened["recommendation"]["vendor"] == "Beta Industrial"
failure = asyncio.run(agent.compare_quotes(
ctx,
"studio-quote-invalid",
[QuoteInput(vendor="Only Vendor", unit_price=10, quantity=1, delivery_days=5, warranty_months=6)],
QuoteWeights(price=50, delivery=30, warranty=20),
))
assert failure["ok"] is False
assert failure["code"] == "at_least_two_quotes_required"
assert "action" in failure
def test_browser_upload_bridge_bounds_and_parses():
payload = json.dumps({"quotes": SUCCESS_ARGS["quotes"]}).encode("utf-8")
parsed = _parse_browser_documents([
BrowserDocument(filename="quotes.json", media_type="application/json", data_base64=base64.b64encode(payload).decode("ascii"))
])
assert parsed["ok"] is True
assert len(parsed["quotes"]) == 2
too_large = base64.b64encode(b"x" * (64 * 1024 + 1)).decode("ascii")
invalid = _parse_browser_documents([
BrowserDocument(filename="too-big.txt", media_type="text/plain", data_base64=too_large)
])
assert invalid["ok"] is False
assert invalid["code"] == "upload_too_large"
bad_b64 = _parse_browser_documents([
BrowserDocument(filename="bad.csv", media_type="text/csv", data_base64="not valid base64")
])
assert bad_b64["ok"] is False
assert bad_b64["code"] == "invalid_base64"
def test_mcp_typed_tool_schemas_are_present():
card = QuoteJudgeStudioV1().card()
schemas = {skill.name: skill.input_schema for skill in card.skills}
compare_schema = schemas["compare_quotes"]
assert set(compare_schema["required"]) == {"comparison_id", "quotes", "weights"}
assert "quotes" in compare_schema["properties"]
assert schemas["upload_quote_file"]["properties"]["file"]["x-a2a-file-upload"]["max_bytes"] == 64 * 1024