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

This commit is contained in:
a2a-cloud
2026-07-18 05:08:52 +00:00
parent 45c8a6ea07
commit 5f950cc458

View File

@@ -1,23 +1,10 @@
from __future__ import annotations
import asyncio
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 QuoteInput, QuoteJudgeStudioV1, QuoteWeights, _build_comparison_result
ROOT = Path(__file__).resolve().parents[1]
SUCCESS_QUOTES = [
QuoteInput(vendor="Acme Bearings", unit_price=10.75, quantity=500, delivery_days=12, warranty_months=12),
QuoteInput(vendor="Beta Industrial", unit_price=9.1, quantity=500, delivery_days=8, warranty_months=18),
]
SUCCESS_WEIGHTS = QuoteWeights(price=50, delivery=30, warranty=20)
def test_full_stack_product_contract():
manifest = (ROOT / "a2a.yaml").read_text(encoding="utf-8")
@@ -25,88 +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 "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
assert "ctx.llm" not in source
assert "Skill runner" not in frontend, "replace the generic scaffold with the product workflow"
forbidden = ["A2A_LITELLM_KEY", "OPENAI_API_KEY", "svc.cluster.local", "DATABASE_URL"]
assert not any(secret in frontend 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_quotes_from_browser_upload",
"compare_quotes_from_upload",
"compare_uploaded_quotes_browser",
"compare_uploaded_quote_file",
}.issubset(skill_names)
assert card.version == "0.1.2"
assert "version: 0.1.2" in (ROOT / "a2a.yaml").read_text(encoding="utf-8")
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"
assert card.runtime.pricing.caller_pays_llm is False
def test_scoring_success_fixture_is_deterministic():
result = _build_comparison_result("studio-quote-v1", SUCCESS_QUOTES, SUCCESS_WEIGHTS)
assert result["ok"] is True
assert result["comparison_id"] == "studio-quote-v1"
assert result["recommendation"]["vendor"] == "Beta Industrial"
beta = result["comparison_table"][0]
assert beta["vendor"] == "Beta Industrial"
assert beta["total_price"] == 4550.0
assert beta["scores"]["weighted_total"] == 100.0
def test_failure_fixture_returns_actionable_validation():
result = _build_comparison_result(
"studio-quote-invalid",
[QuoteInput(vendor="Only Vendor", unit_price=10, quantity=1, delivery_days=5, warranty_months=6)],
SUCCESS_WEIGHTS,
)
assert result["ok"] is False
assert result["code"] == "at_least_two_quotes_required"
assert result["validation_errors"]
def test_success_and_reload_persist_with_receipt():
saved: dict[tuple[str, str], dict] = {}
receipts: list[dict] = []
async def fake_save(tenant_key, comparison_id, result):
saved[(tenant_key, comparison_id)] = result
async def fake_load(tenant_key, comparison_id):
return saved.get((tenant_key, comparison_id))
async def fake_receipt(ctx, tenant_key, comparison_id, tool_name, status, inputs, result):
receipt = {"persisted": True, "receipt_id": f"receipt-{tool_name}-{comparison_id}", "input_hash": "abc"}
receipts.append({"tenant_key": tenant_key, "comparison_id": comparison_id, "tool_name": tool_name, "status": status})
return receipt
async def run():
agent = QuoteJudgeStudioV1()
ctx = LocalRunContext(auth=PlatformUserAuth(sub="user-1", user_id=1, email="u@example.com"))
with patch("agent._save_comparison", fake_save), patch("agent._load_comparison", fake_load), patch("agent._persist_receipt", fake_receipt):
success = await agent.invoke("compare_quotes", ctx, comparison_id="studio-quote-v1", quotes=SUCCESS_QUOTES, weights=SUCCESS_WEIGHTS)
reload = await agent.invoke("get_comparison", ctx, comparison_id="studio-quote-v1")
assert success["ok"] is True
assert success["recommendation"]["vendor"] == "Beta Industrial"
assert success["receipt"]["persisted"] is True
assert reload["ok"] is True
assert reload["recommendation"]["vendor"] == "Beta Industrial"
assert len(receipts) >= 2
asyncio.run(run())