import base64 import importlib.util import sys 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 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 sys.modules[spec.name] = module 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.6" 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 "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 "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.6" 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 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" multiple_dates = module._analyze_contract_text( contract_id="studio-contract-multiple-dates", title="Multiple dates", text=( "This Agreement is effective on 2025-01-01 and renews automatically " "on 2026-12-31 unless either party gives at least 60 days written notice." ), ) assert [d.model_dump(include={"kind", "date"}) for d in multiple_dates.deadlines] == [ {"kind": "renewal", "date": "2026-12-31"}, {"kind": "notice", "date": "2026-11-01"}, ] separate_lines = module._analyze_contract_text( contract_id="studio-contract-separate-lines", title="Separate clauses", text=( "Agreement renews on 2026-12-31\n" "60 days written notice applies to another obligation" ), ) assert [d.model_dump(include={"kind", "date"}) for d in separate_lines.deadlines] == [ {"kind": "renewal", "date": "2026-12-31"}, ] @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"): raise AssertionError("timeouts must be configured in connection options") 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", "") assert "idle_in_transaction_session_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 receipt_payload = __import__("json").loads(store["receipts"][0][-1]) assert "source_text" not in str(receipt_payload) assert receipt_payload["result"]["deadline_count"] == 2 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()