import asyncio import base64 import json from pathlib import Path from unittest.mock import Mock from a2a_pack.auth import PlatformUserAuth from a2a_pack.cli.local import load_local_project from a2a_pack.context import LocalRunContext import agent as agent_module from agent import QuoteJudgeStudioV1 ROOT = Path(__file__).resolve().parents[1] AUTH = PlatformUserAuth(sub="user-123", user_id=123, email="buyer@example.com") def run(coro): return asyncio.run(coro) def ctx(): return LocalRunContext(auth=AUTH, task_id="test-task", caller="pytest") 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: 600" 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 and "compare_quotes" in frontend assert "authorizeUrl" in browser_client and "loginUrl" in browser_client forbidden = ["DATABASE_URL", "A2A_LITELLM", "OPENAI_API_KEY", ".svc.cluster.local"] 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_file"} <= skill_names 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 assert card.version == "0.1.4" def test_success_failure_reload_and_receipt_contract(monkeypatch): comparisons = {} receipts = [] def persist(tenant_key, comparison_id, result, receipt): comparisons[(tenant_key, comparison_id)] = { "result": json.loads(json.dumps(result)), "updated_at": receipt["created_at"], "latest_receipt_id": receipt["receipt_id"], } receipts.append(receipt) return {"ok": True} monkeypatch.setattr(agent_module, "_persist_comparison", persist) monkeypatch.setattr( agent_module, "_load_comparison", lambda tenant_key, comparison_id: comparisons.get((tenant_key, comparison_id)), ) agent = QuoteJudgeStudioV1() quotes = [ {"vendor": "Acme Bearings", "quantity": 500, "unit_price": 10.75, "delivery_days": 12, "warranty_months": 12}, {"vendor": "Beta Industrial", "quantity": 500, "unit_price": 9.1, "delivery_days": 8, "warranty_months": 18}, ] weights = {"price": 50, "delivery": 30, "warranty": 20} success = run(agent.local_invoke("compare_quotes", auth=AUTH, comparison_id="studio-quote-v1", quotes=quotes, weights=weights)) assert success["ok"] is True assert success["comparison_id"] == "studio-quote-v1" assert success["recommendation"]["vendor"] == "Beta Industrial" assert success["receipt"]["receipt_id"].startswith("qjr_") assert success["receipt"]["persisted"] is True assert receipts and receipts[-1]["schema"] == "quotejudge.production_receipt.v1" assert receipts[-1]["agent_version"] == "0.1.4" assert "tenant_key" not in json.dumps(success) 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" assert reload["reloaded"] is True failure = run(agent.local_invoke( "compare_quotes", auth=AUTH, comparison_id="studio-quote-invalid", quotes=[{"vendor": "Only Vendor", "quantity": 1, "unit_price": 10, "delivery_days": 5, "warranty_months": 6}], weights=weights, )) assert failure == {"ok": False, "code": "at_least_two_quotes_required", "comparison_id": "studio-quote-invalid"} def test_browser_upload_bridge_success_and_invalid_base64(monkeypatch): monkeypatch.setattr(agent_module, "_persist_comparison", lambda *_args: {"ok": True}) agent = QuoteJudgeStudioV1() payload = base64.b64encode(json.dumps({ "quotes": [ {"vendor": "Acme Bearings", "quantity": 500, "unit_price": 10.75, "delivery_days": 12, "warranty_months": 12}, {"vendor": "Beta Industrial", "quantity": 500, "unit_price": 9.1, "delivery_days": 8, "warranty_months": 18}, ] }).encode()).decode() result = run(agent.local_invoke( "compare_quotes_from_browser_upload", auth=AUTH, comparison_id="upload-quote-v1", documents=[{"filename": "quotes.json", "media_type": "application/json", "data_base64": payload}], weights={"price": 50, "delivery": 30, "warranty": 20}, )) assert result["ok"] is True assert result["recommendation"]["vendor"] == "Beta Industrial" assert result["source"] == "browser_upload" assert result["parsed_documents"] == [{"filename": "quotes.json", "quotes": 2}] invalid = run(agent.local_invoke( "compare_quotes_from_browser_upload", auth=AUTH, comparison_id="upload-bad", documents=[{"filename": "quotes.json", "media_type": "application/json", "data_base64": "not base64 ***"}], weights={"price": 50, "delivery": 30, "warranty": 20}, )) assert invalid["ok"] is False assert invalid["code"] == "invalid_base64" def test_persistence_fails_closed_without_managed_database(monkeypatch): monkeypatch.delenv("DATABASE_URL", raising=False) receipt = {"receipt_id": "qjr_test", "created_at": "2026-01-01T00:00:00+00:00"} persisted = agent_module._persist_comparison("user:123", "quote-1", {}, receipt) loaded = agent_module._load_comparison("user:123", "quote-1") assert persisted == {"ok": False, "code": "persistence_unavailable"} assert loaded == {"ok": False, "code": "persistence_unavailable"} def test_database_exception_returns_stable_failure_code(monkeypatch): monkeypatch.setenv("DATABASE_URL", "postgresql://example.invalid/db") monkeypatch.setattr(agent_module, "psycopg", Mock()) def unavailable(): raise TimeoutError("database connection timed out") monkeypatch.setattr(agent_module, "_connect", unavailable) persisted = agent_module._persist_comparison( "user:123", "quote-1", {"weights": {}, "quote_count": 2, "recommendation": {"vendor": "Acme"}}, { "receipt_id": "qjr_test", "created_at": "2026-01-01T00:00:00+00:00", "skill_name": "compare_quotes", "status": "ok", "input_hash": "input", "result_hash": "result", }, ) assert persisted == {"ok": False, "code": "persistence_unavailable"} def test_database_connection_has_hard_timeouts(monkeypatch): fake_psycopg = Mock() fake_psycopg.connect.return_value = object() monkeypatch.setattr(agent_module, "psycopg", fake_psycopg) monkeypatch.setenv("DATABASE_URL", "postgresql://example.invalid/db") agent_module._connect() _, kwargs = fake_psycopg.connect.call_args assert kwargs["connect_timeout"] == 5 assert "statement_timeout=10000" in kwargs["options"] assert "lock_timeout=5000" in kwargs["options"] def test_mcp_tool_schemas_are_bounded_and_typed(): card = load_local_project(ROOT).agent_cls().card() skills = {skill.name: skill for skill in card.skills} schema = skills["compare_quotes"].input_schema assert schema["type"] == "object" assert schema["additionalProperties"] is False assert set(schema["required"]) == {"comparison_id", "quotes", "weights"} assert schema["properties"]["comparison_id"]["type"] == "string" assert "quotes" in schema["properties"] assert skills["compare_quotes_from_file"].input_schema["properties"]["document"]["x-a2a-file-upload"]["max_bytes"] == 64 * 1024