88 lines
3.6 KiB
Python
88 lines
3.6 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")
|
|
sip_management = (
|
|
ROOT / "docs/contracts/sip-management.openapi.yaml"
|
|
).read_text(encoding="utf-8")
|
|
self.assertIn("openapi: 3.1.0", executor)
|
|
self.assertIn("openapi: 3.1.0", saas)
|
|
cell_agent = (ROOT / "docs/contracts/cell-agent.openapi.yaml").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("openapi: 3.1.0", sip_management)
|
|
self.assertIn("openapi: 3.1.0", cell_agent)
|
|
self.assertIn("CellManagementMtls", cell_agent)
|
|
self.assertIn("/v1/sip/trunks/{trunk_id}/apply:", cell_agent)
|
|
self.assertIn("SipAdminBearer", sip_management)
|
|
self.assertIn("SaasTrunkReadBearer", sip_management)
|
|
self.assertIn("/admin/v1/trunks/{trunk_id}/publish:", sip_management)
|
|
self.assertIn("/readonly/v1/sip/trunks:", sip_management)
|
|
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_ai_config_contract_and_example(self) -> None:
|
|
schema = read_json(ROOT / "docs/contracts/ai-config.schema.json")
|
|
example = read_json(ROOT / "docs/contracts/examples/agent-version.json")
|
|
self.assertEqual(list(Draft202012Validator(schema).iter_errors(example)), [])
|
|
openapi = (ROOT / "docs/contracts/ai-config.openapi.yaml").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("openapi: 3.1.0", openapi)
|
|
self.assertIn("ai.config.publish", openapi)
|
|
self.assertIn("ai.config.read", openapi)
|
|
self.assertIn("/internal/v1/ai/agent-versions:", openapi)
|
|
self.assertNotIn("api_key", json.dumps(example, ensure_ascii=False))
|
|
|
|
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()
|