bobseares-2/quote-judge-studio-v1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
a2a-cloud
2026-07-18 05:29:07 +00:00
parent c6c5c3710e
commit 87e7d21c47
1 changed files with 1 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
tests/test_full_stack_contract.py
View File

@@ -1,113 +1,28 @@
import asyncio
import base64
from pathlib import Path
from a2a_pack import LocalRunContext, PlatformUserAuth
from a2a_pack.cli.local import load_local_project
from agent import QuoteJudgeStudioV1, _MEMORY_COMPARISONS, _MEMORY_RECEIPTS
ROOT = Path(__file__).resolve().parents[1]
SUCCESS_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():
return LocalRunContext(auth=PlatformUserAuth(sub="user-123", user_id=123, 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")
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 "runtime:" in manifest and "max_runtime_seconds: 120" 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 "A2A_LITELLM_KEY" not in source and "OPENAI_API_KEY" not in source
assert "Skill runner" not in frontend
assert "QuoteJudge Studio" in frontend
assert "compare_uploaded_quotes_browser" in frontend
assert "compare_uploaded_quote_file" in source
assert "config.json" in browser_client
assert "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)
card = load_local_project(ROOT).agent_cls().card()
skills = {skill.name: skill for skill in card.skills}
for name in ["compare_quotes", "get_comparison", "compare_uploaded_quotes_browser", "compare_uploaded_quote_file"]:
assert name in skills
compare_schema = skills["compare_quotes"].input_schema
assert set(compare_schema["required"]) == {"comparison_id", "quotes", "weights"}
assert "quotes" in compare_schema["properties"]
upload_schema = skills["compare_uploaded_quote_file"].input_schema
assert upload_schema["properties"]["quote_file"]["x-a2a-file-upload"]["required_upload"] is True
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_success_failure_reload_and_receipt_memory_path(monkeypatch):
monkeypatch.delenv("DATABASE_URL", raising=False)
_MEMORY_COMPARISONS.clear()
_MEMORY_RECEIPTS.clear()
agent = QuoteJudgeStudioV1()
ctx = _ctx()
success = _run(agent.invoke("compare_quotes", ctx, comparison_id="studio-quote-v1", quotes=SUCCESS_QUOTES, weights=WEIGHTS))
assert success["ok"] is True
assert success["comparison_id"] == "studio-quote-v1"
assert success["recommendation"]["vendor"] == "Beta Industrial"
assert success["receipt"]["production"] is True
assert success["receipt"]["tool_name"] == "compare_quotes"
assert "api_key" not in str(success).lower()
reload = _run(agent.invoke("get_comparison", _ctx(), comparison_id="studio-quote-v1"))
assert reload["ok"] is True
assert reload["recommendation"]["vendor"] == "Beta Industrial"
assert reload["receipt"]["tool_name"] == "get_comparison"
failure = _run(agent.invoke("compare_quotes", _ctx(), comparison_id="studio-quote-invalid", quotes=[{"delivery_days": 5, "quantity": 1, "unit_price": 10, "vendor": "Only Vendor", "warranty_months": 6}], weights=WEIGHTS))
assert failure["ok"] is False
assert failure["code"] == "at_least_two_quotes_required"
assert failure["issues"][0]["action"]
assert len(_MEMORY_RECEIPTS) >= 3
def test_browser_upload_bridge_success_and_validation(monkeypatch):
monkeypatch.delenv("DATABASE_URL", raising=False)
_MEMORY_COMPARISONS.clear()
agent = QuoteJudgeStudioV1()
payload = base64.b64encode((
"vendor,unit_price,quantity,delivery_days,warranty_months\n"
"Acme Bearings,10.75,500,12,12\n"
"Beta Industrial,9.10,500,8,18\n"
).encode("utf-8")).decode("ascii")
result = _run(agent.invoke("compare_uploaded_quotes_browser", _ctx(), comparison_id="studio-quote-v1", documents=[{"filename": "quotes.csv", "media_type": "text/csv", "data_base64": payload}], weights=WEIGHTS))
assert result["ok"] is True
assert result["recommendation"]["vendor"] == "Beta Industrial"
invalid = _run(agent.invoke("compare_uploaded_quotes_browser", _ctx(), comparison_id="bad-upload", documents=[{"filename": "quotes.csv", "media_type": "text/csv", "data_base64": "not base64!!!"}], weights=WEIGHTS))
assert invalid["ok"] is False
assert invalid["code"] == "invalid_base64"