73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
import asyncio
|
|
import json
|
|
import unittest
|
|
from concurrent.futures import Future
|
|
|
|
from app.api.producer import MockMessageProducer
|
|
from app.api.server import IngestionAPIServer
|
|
|
|
|
|
def payload():
|
|
return dict(raw_id="delivery-test", device_id="123", caller_num="+19875551234",
|
|
svc="4G_DATA", rat=6, lac="1", ci="2", up_bytes=10,
|
|
down_bytes=20, epoch_start=1726070000, epoch_end=1726070300)
|
|
|
|
|
|
class PendingProducer(MockMessageProducer):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.pending = []
|
|
|
|
def produce(self, topic, key, value):
|
|
future = Future()
|
|
self.pending.append(future)
|
|
return future
|
|
|
|
|
|
class Writer:
|
|
def __init__(self):
|
|
self.data = b""
|
|
|
|
def write(self, value):
|
|
self.data += value
|
|
|
|
async def drain(self):
|
|
pass
|
|
|
|
|
|
class TestAcknowledgedAPI(unittest.IsolatedAsyncioTestCase):
|
|
async def test_response_waits_for_delivery_and_bounds_outstanding(self):
|
|
producer = PendingProducer()
|
|
server = IngestionAPIServer(producer=producer, queue_max_size=1)
|
|
writer = Writer()
|
|
task = asyncio.create_task(server._handle_raw_cdr_ingestion(writer, json.dumps(payload()).encode()))
|
|
await asyncio.sleep(0.02)
|
|
self.assertEqual(writer.data, b"", "202 must await the broker")
|
|
other = Writer()
|
|
await server._handle_raw_cdr_ingestion(other, json.dumps(payload()).encode())
|
|
self.assertIn(b"429", other.data)
|
|
producer.pending[0].set_result(None)
|
|
await task
|
|
self.assertIn(b"202", writer.data)
|
|
|
|
async def test_missing_fields_wrong_types_and_utf8_return_400(self):
|
|
server = IngestionAPIServer(producer=MockMessageProducer())
|
|
for body in (b"{}", b"\xff", json.dumps(payload() | {"raw_id": 42}).encode(),
|
|
json.dumps(payload() | {"rat": True}).encode()):
|
|
with self.subTest(body=body):
|
|
writer = Writer()
|
|
await server._handle_raw_cdr_ingestion(writer, body)
|
|
self.assertIn(b"400", writer.data)
|
|
|
|
async def test_failed_delivery_returns_503_and_next_request_recovers(self):
|
|
producer = MockMessageProducer()
|
|
server = IngestionAPIServer(producer=producer)
|
|
producer.set_healthy(False)
|
|
writer = Writer()
|
|
await server._handle_raw_cdr_ingestion(writer, json.dumps(payload()).encode())
|
|
self.assertIn(b"503", writer.data)
|
|
producer.set_healthy(True)
|
|
writer = Writer()
|
|
await server._handle_raw_cdr_ingestion(writer, json.dumps(payload()).encode())
|
|
self.assertIn(b"202", writer.data)
|