223 lines
8.1 KiB
Python
223 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import http.client
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from agent_call.core import AgentCallService, InMemoryBroker, iso, utcnow
|
|
from agent_call.http import make_server
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def load_fixture() -> dict[str, Any]:
|
|
try:
|
|
return json.loads(
|
|
(ROOT / "docs/contracts/examples/call.execute.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise AssertionError("fixture unavailable") from exc
|
|
|
|
|
|
class HttpContractTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.previous_tokens = os.environ.get("HTTP_TOKENS")
|
|
os.environ["HTTP_TOKENS"] = json.dumps(
|
|
{
|
|
"local": {
|
|
"tenant_ids": ["tenant-demo"],
|
|
"scopes": [
|
|
"outbound.read",
|
|
"outbound.control",
|
|
"recording.upload",
|
|
"recording.complete",
|
|
],
|
|
},
|
|
"replay": {
|
|
"tenant_ids": ["tenant-demo"],
|
|
"scopes": ["outbound.read", "outbound.replay"],
|
|
},
|
|
}
|
|
)
|
|
self.service = AgentCallService(
|
|
db_path=Path(self.temp.name) / "http.sqlite3",
|
|
object_dir=Path(self.temp.name) / "objects",
|
|
broker=InMemoryBroker(),
|
|
)
|
|
self.service.profile["limits"]["tenant_cps"] = 100
|
|
self.service.profile["limits"]["global_cps"] = 100
|
|
self.server = make_server(self.service, "127.0.0.1", 0)
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
self.port = self.server.server_address[1]
|
|
|
|
def tearDown(self) -> None:
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=2)
|
|
self.service.stop()
|
|
if self.previous_tokens is None:
|
|
os.environ.pop("HTTP_TOKENS", None)
|
|
else:
|
|
os.environ["HTTP_TOKENS"] = self.previous_tokens
|
|
self.temp.cleanup()
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
body: dict[str, Any] | None = None,
|
|
tenant: str | None = "tenant-demo",
|
|
auth_name: str | None = None,
|
|
idem: str | None = None,
|
|
include_auth: bool = True,
|
|
) -> tuple[int, dict[str, Any]]:
|
|
headers = {"X-Request-ID": "http-test"}
|
|
if tenant is not None:
|
|
headers["X-Tenant-ID"] = tenant
|
|
if include_auth and auth_name is None:
|
|
token_map = json.loads(os.environ["HTTP_TOKENS"])
|
|
auth_name = next(iter(token_map), None)
|
|
if include_auth and auth_name is not None:
|
|
headers["Authorization"] = f"Bearer {auth_name}"
|
|
encoded: bytes | None = None
|
|
if body is not None:
|
|
encoded = json.dumps(body, ensure_ascii=False).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
if idem is not None:
|
|
headers["Idempotency-Key"] = idem
|
|
connection = http.client.HTTPConnection("127.0.0.1", self.port, timeout=3)
|
|
try:
|
|
connection.request(method, path, body=encoded, headers=headers)
|
|
response = connection.getresponse()
|
|
raw = response.read()
|
|
status = response.status
|
|
except OSError as exc:
|
|
raise AssertionError("HTTP test server failed") from exc
|
|
finally:
|
|
connection.close()
|
|
if not raw.startswith(b"{"):
|
|
raise AssertionError("HTTP response is not JSON")
|
|
try:
|
|
parsed = json.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise AssertionError("HTTP response is not JSON") from exc
|
|
if not isinstance(parsed, dict):
|
|
raise TypeError("HTTP response is not an object")
|
|
return status, parsed
|
|
|
|
def publish_success(self) -> dict[str, Any]:
|
|
command = copy.deepcopy(load_fixture())
|
|
command["command_id"] = "http_cmd"
|
|
command["trace_id"] = "http_trace"
|
|
command["payload"]["execution_id"] = "http_exec"
|
|
command["issued_at"] = iso(utcnow())
|
|
self.service.publish_execute(command)
|
|
self.service.wait_for_idle()
|
|
return self.service.get_command("tenant-demo", "http_cmd")
|
|
|
|
def test_health_is_public_but_business_queries_are_authenticated(self) -> None:
|
|
status, body = self.request(
|
|
"GET", "/healthz/live", tenant=None, include_auth=False
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(body["status"], "live")
|
|
status, body = self.request(
|
|
"GET", "/internal/v1/outbound/commands/missing", include_auth=False
|
|
)
|
|
self.assertEqual(status, 401)
|
|
self.assertEqual(body["code"], "UNAUTHORIZED")
|
|
|
|
def test_control_query_and_no_http_execute_endpoint(self) -> None:
|
|
snapshot = self.publish_success()
|
|
status, body = self.request(
|
|
"GET", f"/internal/v1/outbound/commands/{snapshot['command_id']}"
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(body["status"], "completed")
|
|
status, body = self.request(
|
|
"POST", "/internal/v1/outbound/commands", body={}, idem="not-an-execute"
|
|
)
|
|
self.assertEqual(status, 404)
|
|
pause = {
|
|
"command_id": "http_pause",
|
|
"action": "pause",
|
|
"expected_task_revision": 1,
|
|
"reason": "test",
|
|
}
|
|
status, body = self.request(
|
|
"POST",
|
|
"/internal/v1/outbound/tasks/task-demo/controls",
|
|
pause,
|
|
idem="http_pause",
|
|
)
|
|
self.assertEqual(status, 202)
|
|
self.service.process_once()
|
|
status, body = self.request("GET", "/internal/v1/outbound/commands/http_pause")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(body["status"], "applied")
|
|
status, body = self.request(
|
|
"GET", "/internal/v1/outbound/commands/http_cmd", tenant="tenant-b"
|
|
)
|
|
self.assertEqual(status, 403)
|
|
|
|
def test_replay_requires_replay_scope(self) -> None:
|
|
snapshot = self.publish_success()
|
|
body = {"command_id": "http_replay", "reason": "retry"}
|
|
status, response = self.request(
|
|
"POST",
|
|
f"/internal/v1/outbound/commands/{snapshot['command_id']}/replays",
|
|
body,
|
|
auth_name="local",
|
|
idem="http_replay",
|
|
)
|
|
self.assertEqual(status, 403)
|
|
self.assertEqual(response["code"], "FORBIDDEN")
|
|
status, response = self.request(
|
|
"POST",
|
|
f"/internal/v1/outbound/commands/{snapshot['command_id']}/replays",
|
|
body,
|
|
auth_name="replay",
|
|
idem="http_replay",
|
|
)
|
|
self.assertEqual(status, 202)
|
|
self.assertEqual(response["status"], "accepted")
|
|
|
|
def test_recording_upload_contract_is_scoped(self) -> None:
|
|
snapshot = self.publish_success()
|
|
call = self.service.get_call("tenant-demo", snapshot["call_id"])
|
|
recording = call["recordings"][0]
|
|
request = {
|
|
"recording_id": recording["recording_id"],
|
|
"call_id": snapshot["call_id"],
|
|
"content_type": "audio/wav",
|
|
"size_bytes": recording["size_bytes"],
|
|
"checksum_algorithm": "SHA-256",
|
|
"checksum": recording["checksum"],
|
|
"channels": recording["channels"],
|
|
"sample_rate_hz": recording["sample_rate_hz"],
|
|
"duration_ms": recording["duration_ms"],
|
|
}
|
|
status, body = self.request(
|
|
"POST",
|
|
"/internal/v1/outbound/recording-uploads",
|
|
request,
|
|
idem="upload-http",
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(body["recording_id"], recording["recording_id"])
|
|
self.assertTrue(body["upload_url"].startswith("/_mock/oss/"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|