153 lines
6.2 KiB
Python
153 lines
6.2 KiB
Python
"""Exercise HTTP framing and broker acknowledgement through real TCP sockets."""
|
|
import asyncio
|
|
import json
|
|
import unittest
|
|
from concurrent.futures import Future
|
|
|
|
from app.api.server import IngestionAPIServer
|
|
|
|
|
|
class PendingProducer:
|
|
def __init__(self):
|
|
self.deliveries = []
|
|
self.produced = asyncio.Event()
|
|
self.closed = False
|
|
|
|
def produce(self, topic, key, value):
|
|
future = Future()
|
|
self.deliveries.append(future)
|
|
self.produced.set()
|
|
return future
|
|
|
|
def is_healthy(self):
|
|
return True
|
|
|
|
def flush(self, timeout=5):
|
|
return sum(not future.done() for future in self.deliveries)
|
|
|
|
def close(self, timeout=5):
|
|
self.closed = True
|
|
|
|
|
|
class HTTPBoundaryTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.producer = PendingProducer()
|
|
self.server = None
|
|
self.writers = []
|
|
self.tasks = []
|
|
|
|
async def asyncTearDown(self):
|
|
for future in self.producer.deliveries:
|
|
if not future.done():
|
|
future.set_result(None)
|
|
for writer in self.writers:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
if self.server is not None:
|
|
await asyncio.wait_for(self.server.stop(), timeout=2)
|
|
for task in self.tasks:
|
|
if not task.done():
|
|
task.cancel()
|
|
await asyncio.gather(*self.tasks, return_exceptions=True)
|
|
|
|
async def start_server(self, **options):
|
|
self.server = IngestionAPIServer(host="127.0.0.1", port=0,
|
|
producer=self.producer, **options)
|
|
await self.server.start()
|
|
self.port = self.server._server.sockets[0].getsockname()[1]
|
|
|
|
async def send(self, request):
|
|
reader, writer = await asyncio.open_connection("127.0.0.1", self.port)
|
|
self.writers.append(writer)
|
|
writer.write(request)
|
|
await writer.drain()
|
|
task = asyncio.create_task(reader.read())
|
|
self.tasks.append(task)
|
|
return task
|
|
|
|
def valid_request(self):
|
|
body = json.dumps(dict(raw_id="http-boundary", device_id="123",
|
|
caller_num="0019875551234", svc="4G_DATA", rat=6,
|
|
up_bytes=100, down_bytes=200, lac="1", ci="2",
|
|
epoch_start=1726070000, epoch_end=1726070300)).encode()
|
|
return (b"POST /api/v1/cdr/raw HTTP/1.1\r\nHost: localhost\r\n"
|
|
+ f"Content-Length: {len(body)}\r\n\r\n".encode() + body)
|
|
|
|
async def assert_response(self, task, status):
|
|
response = await asyncio.wait_for(asyncio.shield(task), timeout=2)
|
|
headers, body = response.split(b"\r\n\r\n", 1)
|
|
self.assertEqual(int(headers.split(b" ")[1]), status, response)
|
|
decoded = json.loads(body)
|
|
if status >= 400:
|
|
self.assertIn("error", decoded)
|
|
return decoded
|
|
|
|
async def test_malformed_framing_returns_structured_400(self):
|
|
await self.start_server()
|
|
headers = [b"Content-Length: invalid", b"Content-Length: -1",
|
|
b"Content-Length: 2\r\nContent-Length: 3",
|
|
b"Transfer-Encoding: chunked",
|
|
b"Transfer-Encoding: chunked\r\nContent-Length: 2"]
|
|
for header in headers:
|
|
with self.subTest(header=header):
|
|
task = await self.send(b"POST /api/v1/cdr/raw HTTP/1.1\r\n" + header + b"\r\n\r\n{}")
|
|
await self.assert_response(task, 400)
|
|
self.assertEqual(self.producer.deliveries, [])
|
|
|
|
async def test_oversized_declared_body_is_rejected_before_reading(self):
|
|
await self.start_server(max_body_bytes=32)
|
|
task = await self.send(b"POST /api/v1/cdr/raw HTTP/1.1\r\nContent-Length: 33\r\n\r\n")
|
|
await self.assert_response(task, 413)
|
|
self.assertEqual(self.producer.deliveries, [])
|
|
|
|
async def test_incomplete_body_times_out_with_structured_408(self):
|
|
await self.start_server(read_timeout=0.03)
|
|
task = await self.send(b"POST /api/v1/cdr/raw HTTP/1.1\r\nContent-Length: 20\r\n\r\n{")
|
|
await self.assert_response(task, 408)
|
|
self.assertEqual(self.producer.deliveries, [])
|
|
|
|
async def test_response_waits_for_broker_acknowledgement(self):
|
|
await self.start_server(delivery_timeout=1)
|
|
task = await self.send(self.valid_request())
|
|
await asyncio.wait_for(self.producer.produced.wait(), timeout=1)
|
|
await asyncio.sleep(0.03)
|
|
self.assertFalse(task.done(), "HTTP accepted before broker acknowledgement")
|
|
self.producer.deliveries[0].set_result(None)
|
|
await self.assert_response(task, 202)
|
|
|
|
async def test_timeout_holds_capacity_until_delivery_finishes(self):
|
|
await self.start_server(queue_max_size=1, delivery_timeout=0.03)
|
|
first = await self.send(self.valid_request())
|
|
await self.assert_response(first, 503)
|
|
self.assertEqual(len(self.producer.deliveries), 1)
|
|
self.assertFalse(self.producer.deliveries[0].cancelled())
|
|
await self.assert_response(await self.send(self.valid_request()), 429)
|
|
self.producer.deliveries[0].set_result(None)
|
|
await asyncio.sleep(0.03)
|
|
self.server.delivery_timeout = 1.0
|
|
self.producer.produced.clear()
|
|
third = await self.send(self.valid_request())
|
|
await asyncio.wait_for(self.producer.produced.wait(), timeout=1)
|
|
self.assertEqual(len(self.producer.deliveries), 2)
|
|
self.producer.deliveries[1].set_result(None)
|
|
await self.assert_response(third, 202)
|
|
|
|
async def test_shutdown_waits_for_outstanding_acknowledgement(self):
|
|
await self.start_server(delivery_timeout=1)
|
|
request = await self.send(self.valid_request())
|
|
await asyncio.wait_for(self.producer.produced.wait(), timeout=1)
|
|
stop = asyncio.create_task(self.server.stop())
|
|
self.tasks.append(stop)
|
|
await asyncio.sleep(0.03)
|
|
self.assertFalse(stop.done(), "Shutdown discarded an outstanding publication")
|
|
self.assertFalse(self.producer.closed)
|
|
self.producer.deliveries[0].set_result(None)
|
|
await asyncio.wait_for(stop, timeout=2)
|
|
await self.assert_response(request, 202)
|
|
self.assertTrue(self.producer.closed)
|
|
self.server = None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|