import asyncio from pathlib import Path from a2a_pack import LocalRunContext, PlatformUserAuth from a2a_pack.cli.local import load_local_project from a2a_pack.mcp import skills_to_tools import agent ROOT = Path(__file__).resolve().parents[1] def _auth() -> PlatformUserAuth: return PlatformUserAuth(sub="test-user", user_id=42, email="user@example.test", scopes=["agent:invoke"]) 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 "version: 0.1.5" in manifest assert "public: false" in manifest assert "frontend:" in manifest and "mount: /app" in manifest assert "resources:" in manifest and "databases:" in manifest assert "scope: user" 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 and "ctx.llm" not in source assert "A2A_LITELLM_KEY" not in source and "OPENAI_API_KEY" not in source assert "SET %s" not in source and "statement_timeout=5000" in source assert "contract_clock_execution_receipts" in source assert "Skill runner" not in frontend assert "ContractClock Studio" in frontend assert "analyze_contract_upload" in frontend assert "config.auth?.authorizeUrl || config.auth?.loginUrl" in browser_client forbidden_frontend = ["DATABASE_URL", "A2A_LITELLM", "OPENAI_API_KEY", ".svc.cluster.local"] assert not any(token in frontend or token in browser_client for token in forbidden_frontend) 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 {"analyze_contract", "get_contract_deadlines", "analyze_contract_file", "analyze_contract_upload"} <= 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" def test_success_and_failure_extraction_are_explicit_date_only(): success = agent._analyze_text( contract_id="studio-contract-v1", title="Studio Renewal Agreement", text=( "This Agreement renews automatically on 2026-12-31 unless either " "party gives at least 60 days written notice. Invoices are due 30 " "days after receipt." ), ) assert success["ok"] is True assert success["contract_id"] == "studio-contract-v1" assert success["deadlines"] == [ {"kind": "renewal", "date": "2026-12-31"}, {"kind": "notice", "date": "2026-11-01"}, ] failure = agent._analyze_text( contract_id="studio-contract-invalid", title="Ambiguous Agreement", text="This Agreement renews sometime next spring unless notice is given well in advance.", ) assert failure == { "ok": False, "code": "ambiguous_date", "contract_id": "studio-contract-invalid", } def test_renewal_date_is_bound_to_the_renewal_clause(): result = agent._analyze_text( contract_id="multi-date-contract", title="Multi-date contract", text=( "This Agreement is effective on 2025-01-01. " "It renews automatically on 2026-12-31 unless either party gives " "at least 60 days written notice." ), ) assert result["ok"] is True assert result["deadlines"] == [ {"kind": "renewal", "date": "2026-12-31"}, {"kind": "notice", "date": "2026-11-01"}, ] def test_browser_upload_validation(): doc = agent.BrowserDocument( filename="contract.txt", media_type="text/plain", data_base64="QWdyZWVtZW50IHJlbmV3cyBvbiAyMDI2LTEyLTMxLg==", ) assert "2026-12-31" in agent._decode_browser_document(doc) bad_doc = agent.BrowserDocument(filename="contract.txt", media_type="text/plain", data_base64="not-base64") try: agent._decode_browser_document(bad_doc) except agent.UploadRejected as exc: assert exc.result["code"] == "invalid_base64" else: # pragma: no cover raise AssertionError("invalid base64 should be rejected") def test_acceptance_success_reload_and_receipt_with_mocked_db(monkeypatch): store = {} async def fake_save_analysis(**kwargs): analysis = dict(kwargs["analysis"]) analysis["receipt"] = { "agent": agent.ContractClockStudioV1.name, "version": agent.ContractClockStudioV1.version, "skill": kwargs["skill_name"], "input_hash": "test-hash", "status": "ok", } store[(kwargs["tenant_key"], kwargs["contract_id"])] = { "title": kwargs["title"], "deadlines": analysis["deadlines"], "receipt": analysis["receipt"], "updated_at": "2026-01-01T00:00:00+00:00", } async def fake_load_contract(**kwargs): return store.get((kwargs["tenant_key"], kwargs["contract_id"])) monkeypatch.setattr(agent, "_save_analysis", fake_save_analysis) monkeypatch.setattr(agent, "_load_contract", fake_load_contract) async def run(): app = agent.ContractClockStudioV1() ctx = LocalRunContext(auth=_auth(), task_id="acceptance-task") result = await app.invoke( "analyze_contract", ctx, contract_id="studio-contract-v1", title="Studio Renewal Agreement", text=( "This Agreement renews automatically on 2026-12-31 unless either " "party gives at least 60 days written notice. Invoices are due 30 days after receipt." ), ) assert result["ok"] is True assert result["deadlines"] == [ {"kind": "renewal", "date": "2026-12-31"}, {"kind": "notice", "date": "2026-11-01"}, ] reload = await app.invoke("get_contract_deadlines", ctx, contract_id="studio-contract-v1") assert reload["ok"] is True assert reload["contract_id"] == "studio-contract-v1" assert reload["deadlines"] == result["deadlines"] assert reload["receipt"]["status"] == "ok" asyncio.run(run()) def test_mcp_tools_are_declared_for_list_and_call_contract(): app = agent.ContractClockStudioV1() tools = skills_to_tools(app) names = {tool["name"] for tool in tools} assert "analyze_contract" in names assert "get_contract_deadlines" in names get_tool = next(tool for tool in tools if tool["name"] == "get_contract_deadlines") assert get_tool["inputSchema"]["properties"]["contract_id"]["type"] == "string"