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()