"""Durability regressions using broker acknowledgement and partition-aware doubles.""" import json import time import unittest from concurrent.futures import Future from types import SimpleNamespace from unittest.mock import Mock, patch from app.worker.consumer import ProcessingWorker from app.worker.metrics import WorkerMetrics from app.worker.normalizer import normalize_cdr_payload def payload(**changes): data = dict(raw_id="regression-1", device_id="310410998877665", caller_num="0019875551234", svc="4G_DATA", rat=6, up_bytes=100, down_bytes=200, lac="4021", ci="8812", epoch_start=1726070000, epoch_end=1726070300, src_ip="10.0.0.1") data.update(changes) return data class Message: def __init__(self, data, offset=0, partition=0): self.data = data if isinstance(data, bytes) else json.dumps(data).encode() self.position = offset self.partition_id = partition def value(self): return self.data def topic(self): return "cdr.raw" def partition(self): return self.partition_id def offset(self): return self.position def timestamp(self): return (1, int((time.time() - 2) * 1000)) def error(self): return None class Consumer: def __init__(self, events, error=None, partition_error=False): self.events = events self.error = error self.partition_error = partition_error self.commits = [] def commit(self, offsets=None, message=None, asynchronous=False): if self.error: raise self.error if offsets is not None: positions = {(item.topic, item.partition): item.offset for item in offsets} elif message is not None: positions = {(message.topic(), message.partition()): message.offset() + 1} else: positions = "implicit-current-position" self.events.append(("commit", positions)) self.commits.append(positions) return [SimpleNamespace(error=RuntimeError("partition commit failed"))] if self.partition_error else [] def close(self): pass class Producer: def __init__(self, events, failures=()): self.events = events self.failures = set(failures) self.messages = [] def produce(self, topic, key, value): self.messages.append(json.loads(value)) result = Future() if len(self.messages) in self.failures: result.set_exception(ConnectionError("broker rejected DLQ record")) else: self.events.append(("dlq_ack", key)) result.set_result(None) return result def close(self): pass def flush(self, timeout=5): return 0 class DeliverySafetyTests(unittest.TestCase): def setUp(self): self.events = [] self.consumer = Consumer(self.events) self.producer = Producer(self.events) self.ch = Mock() self.ch.insert_batch.side_effect = lambda records: self.events.append(("insert", len(records))) or True self.metrics = WorkerMetrics() self.patches = [patch("app.worker.consumer.worker_metrics", self.metrics), patch("app.worker.dlq.worker_metrics", self.metrics)] for item in self.patches: item.start() self.addCleanup(item.stop) self.worker = ProcessingWorker(consumer=self.consumer, dlq_producer=self.producer, clickhouse=self.ch, batch_size=100, internal_server=Mock()) def test_invalid_record_cannot_commit_past_buffered_valid_record(self): self.worker.process_single_message(Message(payload(), offset=0)) self.worker.process_single_message(Message(b"{malformed", offset=1)) self.assertEqual(self.events[0], ("insert", 1)) self.assertEqual(self.consumer.commits[-1], {("cdr.raw", 0): 2}) self.assertEqual(len(self.worker.buffer), 0) def test_batch_commits_only_its_explicit_partition_offsets(self): self.worker.process_single_message(Message(payload(), offset=4, partition=0)) self.worker.process_single_message(Message(payload(raw_id="second"), offset=9, partition=1)) self.worker.flush_buffer() self.assertEqual(self.consumer.commits, [{("cdr.raw", 0): 5, ("cdr.raw", 1): 10}]) def test_failed_dlq_ack_never_commits_or_counts_success(self): self.producer.failures = {1} with self.assertRaises(Exception): self.worker.process_single_message(Message(b"{malformed")) self.assertEqual(self.consumer.commits, []) self.assertEqual(self.metrics.get_internal_stats()["pipeline"]["total_dlq"], 0) def test_partial_batch_dlq_failure_keeps_all_offsets_uncommitted(self): self.ch.insert_batch.side_effect = ConnectionError("ClickHouse unavailable") self.producer.failures = {2} self.worker.process_single_message(Message(payload(), offset=0)) self.worker.process_single_message(Message(payload(raw_id="second"), offset=1)) with self.assertRaises(Exception): self.worker.flush_buffer() self.assertEqual(self.consumer.commits, []) self.assertEqual(self.metrics.get_internal_stats()["pipeline"]["total_dlq"], 1) def test_commit_exception_propagates_without_routing_inserted_batch_to_dlq(self): self.consumer.error = ConnectionError("commit unavailable") self.worker.process_single_message(Message(payload())) with self.assertRaises(Exception): self.worker.flush_buffer() self.ch.insert_batch.assert_called_once() self.assertEqual(self.producer.messages, []) def test_partition_commit_error_is_not_success(self): self.consumer.partition_error = True self.worker.process_single_message(Message(payload())) with self.assertRaises(Exception): self.worker.flush_buffer() self.assertEqual(self.producer.messages, []) def test_clickhouse_failure_dlq_retains_original_vendor_payload(self): self.ch.insert_batch.side_effect = ConnectionError("ClickHouse unavailable") original = payload(vendor_extension={"keep": True}) self.worker.process_single_message(Message(original)) self.worker.flush_buffer() received = self.producer.messages[0]["original_payload"] self.assertEqual(json.loads(received) if isinstance(received, str) else received, original) def test_successful_insert_populates_kafka_latency(self): self.worker.process_single_message(Message(payload())) self.worker.flush_buffer() self.assertIn("cdr_kafka_to_clickhouse_latency_seconds_count 1", self.metrics.get_prometheus_metrics()) class InputBoundaryTests(unittest.TestCase): def test_invalid_values_are_rejected_without_raising(self): cases = [("epoch_start", 10**100), ("epoch_end", 10**100), ("epoch_end", 2**32), ("epoch_start", -1), ("epoch_start", 1726070000.5), ("rat", 6.9), ("rat", True), ("up_bytes", -0.5), ("up_bytes", True), ("up_bytes", 2**64), ("up_bytes", 2**64 - 1), ("src_ip", "999.1.2.3"), ("src_ip", "::1"), ("raw_id", 123)] for field, value in cases: with self.subTest(field=field, value=value): record, reason, failed_field, _ = normalize_cdr_payload(json.dumps(payload(**{field: value})).encode()) self.assertIsNone(record) self.assertIsNotNone(reason) self.assertIsNotNone(failed_field) def test_invalid_utf8_and_nonfinite_json_are_rejected(self): for value in (b"\xff", json.dumps(payload(up_bytes=float("nan"))).encode()): with self.subTest(value=value): record, reason, _, _ = normalize_cdr_payload(value) self.assertIsNone(record) self.assertIsNotNone(reason) def test_clickhouse_rows_use_utc_and_native_ipv4(self): from datetime import timedelta from ipaddress import IPv4Address record, reason, _, _ = normalize_cdr_payload(json.dumps(payload()).encode()) self.assertIsNone(reason) row = record.to_row() self.assertEqual(row[11].utcoffset(), timedelta(0)) self.assertIsInstance(row[13], IPv4Address) def test_prometheus_labels_escape_quotes_backslashes_and_newlines(self): metrics = WorkerMetrics() metrics.record_processed('svc"\\\nname', "LTE_4G") self.assertIn('service_type="svc\\"\\\\\\nname"', metrics.get_prometheus_metrics()) if __name__ == "__main__": unittest.main()