a2a-source-edit: write tests/test_full_stack_contract.py

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()