import asyncio import json import unittest from app.api.producer import MockMessageProducer from app.api.server import IngestionAPIServer class TestIngestionAPIServer(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.mock_producer = MockMessageProducer() self.server = IngestionAPIServer( host="127.0.0.1", port=18000, queue_max_size=2, # small queue to easily test backpressure producer=self.mock_producer, worker_stats_url="http://127.0.0.1:18001/internal/stats", ) await self.server.start() async def asyncTearDown(self): await self.server.stop() async def _http_request( self, method: str, path: str, body: bytes = b"", headers: dict = None ) -> tuple[int, dict, bytes]: reader, writer = await asyncio.open_connection("127.0.0.1", 18000) req_headers = headers or {} req_lines = [f"{method} {path} HTTP/1.1", "Host: 127.0.0.1:18000"] for k, v in req_headers.items(): req_lines.append(f"{k}: {v}") if body and "content-length" not in {k.lower() for k in req_headers}: req_lines.append(f"Content-Length: {len(body)}") req_lines.append("") req_lines.append("") raw_req = "\r\n".join(req_lines).encode("utf-8") + body writer.write(raw_req) await writer.drain() # Read response status line resp_line = await reader.readline() parts = resp_line.decode("utf-8").strip().split(" ", 2) status = int(parts[1]) # Read response headers resp_headers = {} while True: line = await reader.readline() if not line or line in (b"\r\n", b"\n"): break line_str = line.decode("utf-8").strip() if ":" in line_str: k, v = line_str.split(":", 1) resp_headers[k.strip().lower()] = v.strip() # Read response body c_len = int(resp_headers.get("content-length", 0)) if c_len > 0: resp_body = await reader.readexactly(c_len) else: resp_body = await reader.read() writer.close() await writer.wait_closed() return status, resp_headers, resp_body async def test_health_endpoint(self): status, _, body = await self._http_request("GET", "/health") self.assertEqual(status, 200) data = json.loads(body.decode("utf-8")) self.assertEqual(data["status"], "ok") async def test_valid_cdr_ingestion(self): payload = { "raw_id": "raw-us-east-9988221", "src_ip": "10.240.12.1", "device_id": "310410998877665", "caller_num": "0019875551234", "svc": "4G_DATA", "up_bytes": 1048576, "down_bytes": 4194304, "rat": 6, "lac": "4021", "ci": "8812", "epoch_start": 1726070000, "epoch_end": 1726070300, } body = json.dumps(payload).encode("utf-8") status, _, resp_body = await self._http_request( "POST", "/api/v1/cdr/raw", body=body, headers={"Content-Type": "application/json"}, ) self.assertEqual(status, 202) data = json.loads(resp_body.decode("utf-8")) self.assertEqual(data["status"], "accepted") self.assertEqual(data["raw_id"], "raw-us-east-9988221") # Let async worker dispatch to mock producer await asyncio.sleep(0.05) self.assertEqual(len(self.mock_producer.messages), 1) self.assertEqual(self.mock_producer.messages[0]["key"], "raw-us-east-9988221") async def test_malformed_json_returns_400(self): body = b'{"raw_id": "bad-json", ' # Invalid JSON syntax status, _, resp_body = await self._http_request( "POST", "/api/v1/cdr/raw", body=body, headers={"Content-Type": "application/json"}, ) self.assertEqual(status, 400) data = json.loads(resp_body.decode("utf-8")) self.assertEqual(data["error"], "invalid_json") self.assertIn("detail", data) async def test_non_dict_json_returns_400(self): body = json.dumps(["not", "an", "object"]).encode("utf-8") status, _, resp_body = await self._http_request( "POST", "/api/v1/cdr/raw", body=body, headers={"Content-Type": "application/json"}, ) self.assertEqual(status, 400) data = json.loads(resp_body.decode("utf-8")) self.assertEqual(data["error"], "invalid_payload") async def test_backpressure_returns_429_when_queue_full(self): from concurrent.futures import Future from tests.test_api_delivery import payload futures = [asyncio.get_running_loop().create_future() for _ in range(2)] self.server._pending.update(futures) body = json.dumps(payload()).encode() status, _, resp_body = await self._http_request("POST", "/api/v1/cdr/raw", body=body) self.assertEqual(status, 429) self.assertEqual(json.loads(resp_body)["error"], "queue_full") for future in futures: future.set_result(None) self.server._pending.clear() async def test_broker_unavailable_returns_503(self): self.mock_producer.set_healthy(False) from tests.test_api_delivery import payload body = json.dumps(payload()).encode("utf-8") status, _, resp_body = await self._http_request( "POST", "/api/v1/cdr/raw", body=body, headers={"Content-Type": "application/json"}, ) self.assertEqual(status, 503) data = json.loads(resp_body.decode("utf-8")) self.assertEqual(data["error"], "broker_unavailable") async def test_metrics_endpoint(self): status, headers, body = await self._http_request("GET", "/metrics") self.assertEqual(status, 200) content = body.decode("utf-8") self.assertIn("cdr_api_received_total", content) self.assertIn("cdr_api_acknowledged_total", content) async def test_stats_endpoint(self): status, _, body = await self._http_request("GET", "/stats") self.assertEqual(status, 200) data = json.loads(body.decode("utf-8")) self.assertIn("pipeline", data) self.assertIn("breakdown", data) self.assertIn("total_received", data["pipeline"]) self.assertIn("total_acknowledged", data["pipeline"]) async def test_metrics_separated_counters(self): from app.api.metrics import api_metrics received_before = api_metrics.total_received ack_before = api_metrics.total_acknowledged # Send invalid payload (400 Bad Request) status, _, _ = await self._http_request("POST", "/api/v1/cdr/raw", body=b"{}", headers={"Content-Type": "application/json"}) self.assertEqual(status, 400) self.assertEqual(api_metrics.total_received, received_before + 1) self.assertEqual(api_metrics.total_acknowledged, ack_before) # Send valid payload (202 Accepted) from tests.test_api_delivery import payload valid_body = json.dumps(payload()).encode("utf-8") status, _, _ = await self._http_request("POST", "/api/v1/cdr/raw", body=valid_body, headers={"Content-Type": "application/json"}) self.assertEqual(status, 202) self.assertEqual(api_metrics.total_received, received_before + 2) self.assertEqual(api_metrics.total_acknowledged, ack_before + 1) async def test_not_found_returns_404(self): status, _, body = await self._http_request("GET", "/unknown/route") self.assertEqual(status, 404) data = json.loads(body.decode("utf-8")) self.assertEqual(data["error"], "not_found") if __name__ == "__main__": unittest.main()