a2a-source-edit: write tests/test_full_stack_contract.py
This commit is contained in:
a2a-cloud
@@ -1,29 +1,9 @@
|
||||
import asyncio
|
||||
import base64
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from a2a_pack import LocalRunContext, PlatformUserAuth
|
||||
from a2a_pack.cli.local import load_local_project
|
||||
|
||||
from agent import (
|
||||
BrowserDocument,
|
||||
ContractClockStudioV1,
|
||||
extract_timeline,
|
||||
production_receipt_id,
|
||||
)
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SUCCESS_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."
|
||||
FAILURE_TEXT = "This Agreement renews sometime next spring unless notice is given well in advance."
|
||||
|
||||
|
||||
def ctx():
|
||||
return LocalRunContext(
|
||||
auth=PlatformUserAuth(sub="user-123", user_id=123, email="user@example.test", scopes=["agent:invoke"]),
|
||||
caller="tester",
|
||||
)
|
||||
|
||||
|
||||
def test_full_stack_product_contract():
|
||||
@@ -32,157 +12,17 @@ def test_full_stack_product_contract():
|
||||
frontend = (ROOT / "frontend" / "src" / "App.jsx").read_text(encoding="utf-8")
|
||||
|
||||
assert "frontend:" in manifest and "mount: /app" in manifest
|
||||
assert "public: false" in manifest
|
||||
assert "resources:" in manifest and "databases:" in manifest
|
||||
assert "migrations:" 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
|
||||
assert "ContractClock" in frontend and "analyze_contract" in frontend
|
||||
assert "A2A_LITELLM" not in source and "OPENAI_API_KEY" not in source
|
||||
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)
|
||||
|
||||
load_local_project(ROOT) # proves the manifest and entrypoint import cleanly.
|
||||
card = ContractClockStudioV1().card()
|
||||
skill_names = {skill.name for skill in card.skills}
|
||||
assert {"analyze_contract", "get_contract_deadlines", "analyze_contract_browser_upload", "analyze_contract_file_upload"}.issubset(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", "text", "title"}
|
||||
upload_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_contract_file_upload")
|
||||
assert upload_schema["properties"]["document"]["x-a2a-file-upload"]["max_bytes"] == 262144
|
||||
card = load_local_project(ROOT).agent_cls().card()
|
||||
databases = card.runtime.platform_resources.databases
|
||||
assert databases and databases[0].scope == "user"
|
||||
assert databases, "live Agent Card must declare its managed database"
|
||||
assert databases[0].migrations is not None
|
||||
assert databases[0].migrations.path == "db/migrations"
|
||||
assert card.runtime.pricing.caller_pays_llm is False
|
||||
assert ContractClockStudioV1.workspace_access.enabled is True
|
||||
|
||||
|
||||
def test_extractor_success_fixture_exact_dates():
|
||||
timeline = extract_timeline(SUCCESS_TEXT)
|
||||
compact = [{"kind": item.kind, "date": item.date} for item in timeline.deadlines]
|
||||
assert {"kind": "renewal", "date": "2026-12-31"} in compact
|
||||
assert {"kind": "notice", "date": "2026-11-01"} in compact
|
||||
assert not any(item["date"] == "2026-01-01" for item in compact)
|
||||
|
||||
|
||||
def test_extractor_rejects_ambiguous_fixture():
|
||||
try:
|
||||
extract_timeline(FAILURE_TEXT)
|
||||
except Exception as exc:
|
||||
assert getattr(exc, "code", None) == "ambiguous_date"
|
||||
else:
|
||||
raise AssertionError("ambiguous fixture should fail")
|
||||
|
||||
|
||||
def test_acceptance_success_reload_failure_and_receipt_persistence():
|
||||
agent = ContractClockStudioV1()
|
||||
saved = {}
|
||||
receipts = []
|
||||
|
||||
def fake_persist_contract_with_receipt(
|
||||
tenant,
|
||||
contract_id,
|
||||
title,
|
||||
payload,
|
||||
receipt_id,
|
||||
skill_name,
|
||||
status,
|
||||
):
|
||||
assert tenant == "user:123"
|
||||
saved[(tenant, contract_id)] = {
|
||||
"payload": payload,
|
||||
"receipt_id": receipt_id,
|
||||
"updated_at": "2026-01-01T00:00:00+00:00",
|
||||
}
|
||||
receipts.append((tenant, receipt_id, skill_name, contract_id, status))
|
||||
|
||||
def fake_load_contract(tenant, contract_id):
|
||||
return saved.get((tenant, contract_id))
|
||||
|
||||
with patch(
|
||||
"agent.persist_contract_with_receipt",
|
||||
side_effect=fake_persist_contract_with_receipt,
|
||||
), patch("agent.load_contract", side_effect=fake_load_contract):
|
||||
result = asyncio.run(agent.analyze_contract(ctx(), contract_id="studio-contract-v1", text=SUCCESS_TEXT, title="Studio Renewal Agreement"))
|
||||
assert result.ok is True
|
||||
assert result.persisted is True
|
||||
assert result.contract_id == "studio-contract-v1"
|
||||
assert [{"kind": d.kind, "date": d.date} for d in result.deadlines[:2]] == [
|
||||
{"kind": "renewal", "date": "2026-12-31"},
|
||||
{"kind": "notice", "date": "2026-11-01"},
|
||||
]
|
||||
assert result.receipt_id
|
||||
assert receipts and receipts[0][2] == "analyze_contract"
|
||||
|
||||
reload = asyncio.run(agent.get_contract_deadlines(ctx(), contract_id="studio-contract-v1"))
|
||||
assert reload.ok is True
|
||||
assert reload.receipt_id == result.receipt_id
|
||||
assert [{"kind": d.kind, "date": d.date} for d in reload.deadlines[:2]] == [
|
||||
{"kind": "renewal", "date": "2026-12-31"},
|
||||
{"kind": "notice", "date": "2026-11-01"},
|
||||
]
|
||||
|
||||
failure = asyncio.run(agent.analyze_contract(ctx(), contract_id="studio-contract-invalid", text=FAILURE_TEXT, title="Ambiguous Agreement"))
|
||||
assert failure.ok is False
|
||||
assert failure.code == "ambiguous_date"
|
||||
|
||||
|
||||
def test_browser_upload_bridge_bounds_and_reuses_workflow():
|
||||
agent = ContractClockStudioV1()
|
||||
encoded = base64.b64encode(SUCCESS_TEXT.encode("utf-8")).decode("ascii")
|
||||
with patch("agent.persist_contract_with_receipt", return_value=None):
|
||||
result = asyncio.run(
|
||||
agent.analyze_contract_browser_upload(
|
||||
ctx(),
|
||||
contract_id="studio-contract-v1",
|
||||
title="Studio Renewal Agreement",
|
||||
document=BrowserDocument(filename="contract.txt", media_type="text/plain", data_base64=encoded),
|
||||
)
|
||||
)
|
||||
assert result.ok is True
|
||||
|
||||
bad = asyncio.run(
|
||||
agent.analyze_contract_browser_upload(
|
||||
ctx(),
|
||||
contract_id="studio-contract-v1",
|
||||
title="Studio Renewal Agreement",
|
||||
document=BrowserDocument(filename="contract.txt", media_type="text/plain", data_base64="not base64***"),
|
||||
)
|
||||
)
|
||||
assert bad.ok is False and bad.code == "invalid_base64"
|
||||
|
||||
|
||||
def test_tools_return_structured_auth_failure_for_missing_principal():
|
||||
anonymous = LocalRunContext(
|
||||
auth=PlatformUserAuth(sub=""),
|
||||
caller="tester",
|
||||
)
|
||||
agent = ContractClockStudioV1()
|
||||
|
||||
analyzed = asyncio.run(
|
||||
agent.analyze_contract(
|
||||
anonymous,
|
||||
contract_id="studio-contract-v1",
|
||||
text=SUCCESS_TEXT,
|
||||
title="Studio Renewal Agreement",
|
||||
)
|
||||
)
|
||||
loaded = asyncio.run(
|
||||
agent.get_contract_deadlines(anonymous, contract_id="studio-contract-v1")
|
||||
)
|
||||
listed = asyncio.run(agent.list_contracts(anonymous))
|
||||
|
||||
assert analyzed.ok is False and analyzed.code == "auth_required"
|
||||
assert loaded.ok is False and loaded.code == "auth_required"
|
||||
assert listed.ok is False and listed.code == "auth_required"
|
||||
|
||||
|
||||
def test_receipt_id_is_deterministic_without_secrets():
|
||||
payload = {"contract_id": "studio-contract-v1", "deadlines": [{"kind": "renewal", "date": "2026-12-31"}]}
|
||||
receipt = production_receipt_id("user:123", "analyze_contract", payload)
|
||||
assert receipt == production_receipt_id("user:123", "analyze_contract", payload)
|
||||
assert receipt.startswith("ccr_")
|
||||
assert "DATABASE" not in json.dumps(payload)
|
||||
|
||||
Reference in New Issue
Block a user