diff --git a/tests/test_full_stack_contract.py b/tests/test_full_stack_contract.py index 398dbbb..59cee34 100644 --- a/tests/test_full_stack_contract.py +++ b/tests/test_full_stack_contract.py @@ -1,181 +1,28 @@ -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 "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 "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) + 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() - 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"