telecom-cdr-pipeline/tests/integration/test_ingestion_api.py

from collections.abc import Mapping
from typing import Any

import pytest
from fastapi.testclient import TestClient

from core.api.main import create_app
from core.api.publisher import (
    IngestionQueueFull,
    KafkaPublishError,
)


class FakePublisher:
    def __init__(self) -> None:
        self.messages: list[
            tuple[str, Mapping[str, Any]]
        ] = []

        self.failure: Exception | None = None

    async def start(self) -> None:
        pass

    async def stop(self) -> None:
        pass

    async def publish(
        self,
        payload: Mapping[str, Any],
        key: str,
    ) -> None:
        if self.failure is not None:
            raise self.failure

        self.messages.append(
            (key, payload)
        )


class FakeWorkerStatsClient:
    async def get_stats(
        self,
    ) -> dict[str, Any]:
        return {
            "pipeline": {
                "total_processed": 10,
                "total_dlq": 2,
                "in_flight_buffer": 3,
                "last_flush_at":
                    "2026-09-20T12:00:00Z",
            },
            "breakdown": {
                "by_service_type": {
                    "4G_DATA": 10,
                },
                "by_radio_access_type": {
                    "LTE_4G": 10,
                },
                "dlq_by_reason": {
                    "invalid_rat_value": 2,
                },
            },
        }

    async def close(self) -> None:
        pass


@pytest.fixture
def publisher() -> FakePublisher:
    return FakePublisher()


@pytest.fixture
def client(
    publisher: FakePublisher,
):
    app = create_app(
        publisher=publisher,
        worker_stats_client=FakeWorkerStatsClient(),
    )

    with TestClient(app) as test_client:
        yield test_client


def valid_payload() -> dict:
    return {
        "raw_id": "raw-test-001",
        "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": 1789900000,
        "epoch_end": 1789900300,
    }


def test_health_endpoint(client: TestClient) -> None:
    response = client.get("/health")

    assert response.status_code == 200
    assert response.json() == {
        "status": "ok",
    }


def test_valid_cdr_returns_202(
    client: TestClient,
    publisher: FakePublisher,
) -> None:
    payload = valid_payload()

    response = client.post(
        "/api/v1/cdr/raw",
        json=payload,
    )

    assert response.status_code == 202

    assert response.json() == {
        "status": "accepted",
        "raw_id": "raw-test-001",
    }

    assert len(publisher.messages) == 1

    key, published_payload = publisher.messages[0]

    assert key == "raw-test-001"
    assert published_payload == payload


def test_missing_required_field_returns_400(client: TestClient) -> None:
    payload = valid_payload()
    del payload["caller_num"]

    response = client.post(
        "/api/v1/cdr/raw",
        json=payload,
    )

    assert response.status_code == 400

    body = response.json()

    assert body["error"] == "missing_required_field"
    assert body["failed_field"] == "caller_num"


def test_invalid_rat_returns_400(client: TestClient) -> None:
    payload = valid_payload()
    payload["rat"] = 99

    response = client.post(
        "/api/v1/cdr/raw",
        json=payload,
    )

    assert response.status_code == 400

    body = response.json()

    assert body["error"] == "invalid_rat_value"
    assert body["failed_field"] == "rat"


def test_invalid_epoch_returns_400(client: TestClient) -> None:
    payload = valid_payload()

    payload["epoch_start"] = 200
    payload["epoch_end"] = 100

    response = client.post(
        "/api/v1/cdr/raw",
        json=payload,
    )

    assert response.status_code == 400

    body = response.json()

    assert body["error"] == "invalid_epoch"


def test_malformed_json_returns_400(client: TestClient) -> None:
    response = client.post(
        "/api/v1/cdr/raw",
        content='{"raw_id": invalid}',
        headers={
            "Content-Type": "application/json",
        },
    )

    assert response.status_code == 400

    body = response.json()

    assert body["error"] == "invalid_json"


def test_non_object_json_returns_400(client: TestClient) -> None:
    response = client.post(
        "/api/v1/cdr/raw",
        json=[
            "not",
            "a",
            "cdr",
        ],
    )

    assert response.status_code == 400

    body = response.json()

    assert body["error"] == "invalid_payload"


def test_queue_full_returns_429(
    client: TestClient,
    publisher: FakePublisher,
) -> None:
    publisher.failure = IngestionQueueFull(
        "Ingestion queue is full."
    )

    response = client.post(
        "/api/v1/cdr/raw",
        json=valid_payload(),
    )

    assert response.status_code == 429

    assert response.json()["error"] == "backpressure"


def test_kafka_failure_returns_503(
    client: TestClient,
    publisher: FakePublisher,
) -> None:
    publisher.failure = KafkaPublishError(
        "Kafka unavailable."
    )

    response = client.post(
        "/api/v1/cdr/raw",
        json=valid_payload(),
    )

    assert response.status_code == 503

    assert response.json()["error"] == "kafka_unavailable"


def test_metrics_endpoint(
    client: TestClient,
) -> None:
    response = client.get(
        "/metrics"
    )

    assert response.status_code == 200

    assert (
        "cdr_api_received_total"
        in response.text
    )


def test_stats_endpoint(
    client: TestClient,
) -> None:
    response = client.get(
        "/stats"
    )

    assert response.status_code == 200

    body = response.json()

    assert (
        body["pipeline"]["total_processed"]
        == 10
    )

    assert (
        body["pipeline"]["total_dlq"]
        == 2
    )

    assert (
        body["pipeline"]["in_flight_buffer"]
        == 3
    )

    assert (
        body["breakdown"]
        ["by_service_type"]
        ["4G_DATA"]
        == 10
    )


def test_successful_request_increments_api_metric(
    client: TestClient,
) -> None:
    response = client.post(
        "/api/v1/cdr/raw",
        json=valid_payload(),
    )

    assert response.status_code == 202

    metrics = client.get(
        "/metrics"
    )

    assert (
        'cdr_api_received_total 1.0'
        in metrics.text
    )