a2a-source-edit: write tests/test_full_stack_contract.py
This commit is contained in:
a2a-cloud
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user