260 lines
11 KiB
Python
260 lines
11 KiB
Python
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.8" 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.8"
|
|
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"
|
|
|
|
reconciliation = (ROOT / "db" / "migrations" / "002_reconcile_execution_receipts.sql").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
assert "ADD COLUMN IF NOT EXISTS tool_name" in reconciliation
|
|
assert "COALESCE(tool_name, skill_name)" in reconciliation
|
|
assert "ALTER COLUMN skill_name DROP NOT NULL" in reconciliation
|
|
assert "ALTER COLUMN input_hash SET NOT NULL" in reconciliation
|
|
|
|
legacy_constraint_fix = (
|
|
ROOT / "db" / "migrations" / "003_relax_legacy_receipt_skill.sql"
|
|
).read_text(encoding="utf-8")
|
|
assert "column_name = 'skill_name'" in legacy_constraint_fix
|
|
assert "ALTER COLUMN skill_name DROP NOT NULL" in legacy_constraint_fix
|
|
|
|
|
|
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()
|