From 567df733f5db2290aa8b857a555648ba28c61cb2 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 08:11:39 +0000 Subject: [PATCH] a2a-source-edit: write tests/test_full_stack_contract.py --- tests/test_full_stack_contract.py | 195 +++++++++++++++++++++++++++++- 1 file changed, 191 insertions(+), 4 deletions(-) diff --git a/tests/test_full_stack_contract.py b/tests/test_full_stack_contract.py index 59cee34..9688b1e 100644 --- a/tests/test_full_stack_contract.py +++ b/tests/test_full_stack_contract.py @@ -1,28 +1,215 @@ +import base64 +import importlib.util +import os from pathlib import Path +import pytest +from a2a_pack import LocalRunContext, PlatformUserAuth from a2a_pack.cli.local import load_local_project ROOT = Path(__file__).resolve().parents[1] -def test_full_stack_product_contract(): +def load_agent_module(): + spec = importlib.util.spec_from_file_location("contract_clock_agent", ROOT / "agent.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +def test_full_stack_product_contract_card_and_manifest(): 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") + a2a_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 "migrations:" in manifest and "db/migrations" in manifest + assert "scope: user" in manifest and "db/migrations" in manifest assert "PlatformUserAuth" in source assert "AgentPlatformResources" in source and "AgentDatabase(" in source - assert "Skill runner" not in frontend, "replace the generic scaffold with the product workflow" + 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 "Skill runner" not in frontend + assert "ContractClock" in frontend and "analyze_contract" in frontend + assert "/app/config.json" in a2a_client or "./config.json" in a2a_client + assert "DATABASE_URL" not in 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() + assert card.version == "0.1.5" + skill_names = {skill.name for skill in card.skills} + assert {"analyze_contract", "get_contract_deadlines", "analyze_contract_upload_base64", "analyze_contract_file"} <= skill_names + analyze_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_contract") + assert set(analyze_schema["required"]) == {"contract_id", "title", "text"} + assert analyze_schema["properties"]["text"]["maxLength"] == 80000 + upload_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_contract_file") + assert upload_schema["properties"]["document"]["x-a2a-file-upload"]["max_bytes"] == 512000 databases = card.runtime.platform_resources.databases - assert databases, "live Agent Card must declare its managed database" + assert databases and databases[0].scope == "user" assert databases[0].migrations is not None assert databases[0].migrations.path == "db/migrations" + assert card.mcp_endpoint == "/mcp" + + +def test_explicit_date_only_success_and_failure(): + module = load_agent_module() + result = module._analyze_contract_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 result.ok is True + assert result.contract_id == "studio-contract-v1" + assert [d.model_dump(include={"kind", "date"}) for d in result.deadlines] == [ + {"kind": "renewal", "date": "2026-12-31"}, + {"kind": "notice", "date": "2026-11-01"}, + ] + + bad = module._analyze_contract_text( + contract_id="studio-contract-invalid", + title="Ambiguous Agreement", + text="This Agreement renews sometime next spring unless notice is given well in advance.", + ) + assert bad.ok is False + assert bad.code == "ambiguous_date" + + +@pytest.mark.asyncio +async def test_success_failure_upload_and_reload_with_fake_postgres(monkeypatch): + module = load_agent_module() + store = {"contracts": {}, "deadlines": {}, "receipts": []} + + class FakeCursor: + def __init__(self): + self.last = None + self.rows = [] + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def execute(self, sql, params=None): + params = params or () + normalized = " ".join(sql.split()) + if normalized.startswith("SET"): + return + if "INSERT INTO contracts" in normalized: + tenant, contract_id, title, source_hash = params + key = (tenant, contract_id) + contract = store["contracts"].setdefault(key, {"id": len(store["contracts"]) + 1}) + contract.update({"tenant_key": tenant, "contract_id": contract_id, "title": title, "source_text_hash": source_hash}) + self.last = {"id": contract["id"]} + return + if "DELETE FROM contract_deadlines" in normalized: + store["deadlines"][params[0]] = [] + return + if "INSERT INTO contract_deadlines" in normalized: + contract_pk, kind, deadline_date, source_text, rationale = params + store["deadlines"].setdefault(contract_pk, []).append( + {"kind": kind, "deadline_date": module.date.fromisoformat(deadline_date), "source_text": source_text, "rationale": rationale} + ) + return + if "INSERT INTO execution_receipts" in normalized: + store["receipts"].append(params) + return + if normalized.startswith("SELECT id, title, updated_at FROM contracts"): + tenant, contract_id = params + contract = store["contracts"].get((tenant, contract_id)) + self.last = None if contract is None else {"id": contract["id"], "title": contract["title"], "updated_at": module.datetime(2026, 1, 1, tzinfo=module.timezone.utc)} + return + if normalized.startswith("SELECT kind, deadline_date, source_text, rationale FROM contract_deadlines"): + self.rows = list(store["deadlines"].get(params[0], [])) + return + raise AssertionError(f"unhandled SQL: {normalized}") + + def fetchone(self): + return self.last + + def fetchall(self): + return self.rows + + class FakeConnection: + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def transaction(self): + return self + + def cursor(self): + return FakeCursor() + + class FakePsycopg: + def connect(self, *args, **kwargs): + assert "statement_timeout=5000" in kwargs.get("options", "") + return FakeConnection() + + monkeypatch.setenv("DATABASE_URL", "postgresql://example.invalid/db") + monkeypatch.setitem(__import__("sys").modules, "psycopg", FakePsycopg()) + monkeypatch.setitem(__import__("sys").modules, "psycopg.rows", type("Rows", (), {"dict_row": object()})()) + + agent = module.ContractClockStudioV1() + ctx = LocalRunContext(auth=PlatformUserAuth(sub="u-123", user_id=123, email="u@example.test"), task_id="acceptance") + success = await agent.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 success.ok is True + assert success.receipt_id and success.saved_at + assert len(store["receipts"]) == 1 + + reload_result = await agent.invoke("get_contract_deadlines", ctx, contract_id="studio-contract-v1") + assert reload_result.ok is True + assert [d.model_dump(include={"kind", "date"}) for d in reload_result.deadlines] == [ + {"kind": "renewal", "date": "2026-12-31"}, + {"kind": "notice", "date": "2026-11-01"}, + ] + + failure = await agent.invoke( + "analyze_contract", + ctx, + 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 is False and failure.code == "ambiguous_date" + assert len(store["receipts"]) == 2 + + uploaded = await agent.invoke( + "analyze_contract_upload_base64", + ctx, + contract_id="studio-contract-upload", + title="Uploaded Agreement", + document={ + "filename": "agreement.txt", + "media_type": "text/plain", + "data_base64": base64.b64encode(b"Agreement renews automatically on 2026-12-31 unless either party gives 60 days written notice.").decode("ascii"), + }, + ) + assert uploaded.ok is True + + +def test_structured_auth_failure_without_secrets(): + module = load_agent_module() + result = module._auth_required("studio-contract-v1") + payload = result.model_dump(mode="json") + assert payload["ok"] is False + assert payload["code"] == "auth_required" + assert "secret" not in str(payload).lower()