Files
agent-call/tests/test_contracts.py
T

61 lines
2.2 KiB
Python

from __future__ import annotations
import json
import unittest
from pathlib import Path
from jsonschema import Draft202012Validator
ROOT = Path(__file__).resolve().parents[1]
def read_json(path: Path) -> dict:
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise AssertionError(f"invalid JSON fixture: {path}") from exc
class ContractTests(unittest.TestCase):
def test_openapi_and_mq_contract_inventory(self) -> None:
executor = (ROOT / "docs/contracts/executor.openapi.yaml").read_text(
encoding="utf-8"
)
saas = (ROOT / "docs/contracts/saas.openapi.yaml").read_text(encoding="utf-8")
self.assertIn("openapi: 3.1.0", executor)
self.assertIn("openapi: 3.1.0", saas)
self.assertNotIn("call.execute", executor)
self.assertIn("/internal/v1/outbound/tasks/{task_id}/controls:", executor)
self.assertIn(
"/internal/v1/outbound/commands/{source_command_id}/replays:", executor
)
self.assertIn("/internal/v1/outbound/recording-uploads:", saas)
self.assertIn(
"/internal/v1/outbound/recording-uploads/{upload_id}/complete:", saas
)
def test_command_fixture_and_invalid_version(self) -> None:
schema = read_json(ROOT / "docs/contracts/mq.schema.json")
fixture = read_json(ROOT / "docs/contracts/examples/call.execute.json")
validator = Draft202012Validator(schema)
self.assertEqual(list(validator.iter_errors(fixture)), [])
invalid = dict(fixture)
invalid["schema_version"] = "2.0"
self.assertTrue(list(validator.iter_errors(invalid)))
def test_raw_unicode_tenant_key_and_user_selected_callee_are_preserved(
self,
) -> None:
fixture = read_json(ROOT / "docs/contracts/examples/call.execute.json")
self.assertEqual(fixture["payload"]["callee"], "18601013734")
fixture["tenant_key"] = "租户#原值"
self.assertEqual(fixture["tenant_key"], "租户#原值")
self.assertEqual(
json.loads(json.dumps(fixture, ensure_ascii=False))["tenant_key"],
"租户#原值",
)
if __name__ == "__main__":
unittest.main()