cmvr_edge_ai/tests/unit/test_http_json_sink.py

from __future__ import annotations

import asyncio
import base64
from copy import deepcopy
from dataclasses import dataclass
import json
import logging
from typing import Any

import pytest

import cmvr_edge_ai.connectors.platform.http_json as http_json
from cmvr_edge_ai.connectors.platform.http_json import PlatformHttpJsonSink
from cmvr_edge_ai.contracts import (
    BoundingBox,
    Detection,
    DetectionAlert,
    DetectionResult,
    EncodedImage,
    ImageFrame,
)
from cmvr_edge_ai.core.component import ComponentContext
from cmvr_edge_ai.core.envelope import Envelope
from cmvr_edge_ai.transports.http import HttpClientPool, HttpClientSettings


class FakeHttpStatusError(RuntimeError):
    def __init__(self, status_code: int) -> None:
        self.status_code = status_code
        super().__init__(f"HTTP status {status_code}")


@dataclass
class FakeResponse:
    status_code: int

    def raise_for_status(self) -> None:
        if self.status_code >= 400:
            raise FakeHttpStatusError(self.status_code)


class FakeHttpClient:
    def __init__(self, outcomes: list[FakeResponse | BaseException]) -> None:
        self._outcomes = list(outcomes)
        self.calls: list[dict[str, Any]] = []

    async def post(
        self,
        path: str,
        *,
        json: dict[str, Any],
        headers: dict[str, str],
    ) -> FakeResponse:
        self.calls.append(
            {
                "path": path,
                "json": deepcopy(json),
                "body_id": id(json),
                "headers": dict(headers),
            }
        )
        if not self._outcomes:
            raise AssertionError("fake HTTP client has no configured outcome")
        outcome = self._outcomes.pop(0)
        if isinstance(outcome, BaseException):
            raise outcome
        return outcome


class FakeHttpClientPool(HttpClientPool):
    def __init__(self, client: FakeHttpClient) -> None:
        self.client = client
        self.get_calls: list[tuple[str, HttpClientSettings]] = []

    async def get(
        self, endpoint_id: str, settings: HttpClientSettings
    ) -> FakeHttpClient:
        self.get_calls.append((endpoint_id, settings))
        return self.client


def _alert(
    event_id: str = "ppe-event-123",
    *,
    image: EncodedImage | None = None,
) -> DetectionAlert:
    detection = Detection(
        label="No-Helmet",
        confidence=0.94,
        box=BoundingBox(x_min=10.0, y_min=20.0, x_max=100.0, y_max=200.0),
    )
    return DetectionAlert(
        event_id=event_id,
        rule_id="no-helmet-rule",
        model_id="construction-ppe@1",
        model_name="Construction PPE",
        labels=("No-Helmet",),
        scope="source",
        scope_id="right-camera",
        hit_count=3,
        window_ms=2_000.0,
        first_seen_ns=1_000_000_000,
        last_seen_ns=2_000_000_000,
        triggered_at_ns=2_000_000_000,
        max_confidence=0.94,
        detections=(detection,),
        image=image,
    )


def _envelope(payload: Any, *, trace_id: str = "trace-fallback-456") -> Envelope[Any]:
    return Envelope(
        payload=payload,
        schema_name=(
            "DetectionAlert" if isinstance(payload, DetectionAlert) else "opaque"
        ),
        schema_version=1,
        source_id="right-camera",
        sequence=17,
        captured_at_ns=2_000_000_000,
        received_at_ns=2_010_000_000,
        trace_id=trace_id,
        attributes={"site": "park-a"},
    )


async def _setup_sink(
    client: FakeHttpClient,
    *,
    context_endpoints: dict[str, dict[str, Any]] | None = None,
    **params: Any,
) -> PlatformHttpJsonSink:
    sink = PlatformHttpJsonSink(
        "platform",
        {
            "endpoint": "platform",
            "path": "/v1/detection-alerts",
            **params,
        },
    )
    pool = FakeHttpClientPool(client)
    endpoints = context_endpoints or {
        "platform": {
            "transport": "http",
            "base_url": "http://platform.invalid",
            "timeout_s": 1.0,
            "metadata": {},
            "options": {},
        },
        "cmvr_es": {
            "transport": "grpc",
            "target": "192.168.0.119:50052",
        },
    }
    context = ComponentContext(
        pipeline_id="detection",
        node_id="platform",
        shutdown_event=asyncio.Event(),
        metadata={
            "endpoints": endpoints,
            "http_client_pool": pool,
        },
    )
    await sink.setup(context)
    assert pool.get_calls[0][0] == "platform"
    return sink


def test_detection_alert_event_id_is_the_idempotency_key() -> None:
    async def scenario() -> None:
        client = FakeHttpClient([FakeResponse(200)])
        sink = await _setup_sink(client, max_attempts=1)

        await sink.consume(_envelope(_alert()), input_port="alerts")

        assert len(client.calls) == 1
        call = client.calls[0]
        assert call["headers"] == {"Idempotency-Key": "ppe-event-123"}
        assert call["path"] == "/v1/detection-alerts"
        assert call["json"]["input_port"] == "alerts"
        assert call["json"]["payload"]["event_id"] == "ppe-event-123"
        assert call["json"]["payload"]["image"] is None
        assert "grpc_ip" not in call["json"]

    asyncio.run(scenario())


def test_configured_grpc_endpoint_ip_is_added_to_the_envelope_json() -> None:
    async def scenario() -> None:
        client = FakeHttpClient([FakeResponse(200)])
        sink = await _setup_sink(
            client,
            grpc_endpoint="cmvr_es",
            max_attempts=1,
        )

        await sink.consume(_envelope(_alert()), input_port="alerts")

        body = client.calls[0]["json"]
        assert body["grpc_ip"] == "192.168.0.119"
        assert body["source_id"] == "right-camera"
        assert body["payload"]["event_id"] == "ppe-event-123"

    asyncio.run(scenario())


@pytest.mark.parametrize(
    ("target", "expected_ip"),
    [
        ("10.20.30.40:50052", "10.20.30.40"),
        ("ipv4:10.20.30.40:50052", "10.20.30.40"),
        ("dns:///10.20.30.40:50052", "10.20.30.40"),
        ("[2001:db8::10]:50052", "2001:db8::10"),
        ("ipv6:[2001:db8::10]:50052", "2001:db8::10"),
    ],
)
def test_grpc_target_ip_parser_accepts_literal_ip_targets(
    target: str,
    expected_ip: str,
) -> None:
    assert http_json._grpc_target_ip(target) == expected_ip


@pytest.mark.parametrize(
    "target",
    [
        "cmvr-es.local:50052",
        "192.168.0.119",
        "192.168.0.119:not-a-port",
        "192.168.0.119:0",
        "192.168.0.119:65536",
        "2001:db8::10:50052",
        "unix:///run/cmvr-es.sock",
        "ipv4:[2001:db8::10]:50052",
        "ipv6:10.20.30.40:50052",
    ],
)
def test_grpc_target_ip_parser_rejects_non_literal_or_ambiguous_targets(
    target: str,
) -> None:
    with pytest.raises(ValueError, match="must contain one literal IP and port"):
        http_json._grpc_target_ip(target)


@pytest.mark.parametrize(
    ("grpc_endpoint", "endpoint_config", "message"),
    [
        ("missing", None, "unknown endpoint"),
        (
            "not_grpc",
            {"transport": "http", "base_url": "http://example.invalid"},
            "must use grpc transport",
        ),
        (
            "bad_target",
            {"transport": "grpc", "target": "cmvr-es.local:50052"},
            "must contain one literal IP and port",
        ),
    ],
)
def test_grpc_endpoint_configuration_is_checked_during_setup(
    grpc_endpoint: str,
    endpoint_config: dict[str, Any] | None,
    message: str,
) -> None:
    async def scenario() -> None:
        endpoints: dict[str, dict[str, Any]] = {
            "platform": {
                "transport": "http",
                "base_url": "http://platform.invalid",
                "timeout_s": 1.0,
                "metadata": {},
                "options": {},
            }
        }
        if endpoint_config is not None:
            endpoints[grpc_endpoint] = endpoint_config
        with pytest.raises(ValueError, match=message):
            await _setup_sink(
                FakeHttpClient([FakeResponse(200)]),
                context_endpoints=endpoints,
                grpc_endpoint=grpc_endpoint,
            )

    asyncio.run(scenario())


def test_detection_alert_image_has_a_flat_base64_wire_shape() -> None:
    async def scenario() -> None:
        jpeg = b"\xff\xd8\xff\xe0annotated-image\xff\xd9"
        image = EncodedImage(
            data=jpeg,
            media_type="image/jpeg",
            width=1280,
            height=720,
        )
        client = FakeHttpClient([FakeResponse(200)])
        sink = await _setup_sink(client, max_attempts=1)

        await sink.consume(_envelope(_alert(image=image)), input_port="alerts")

        body = client.calls[0]["json"]
        payload = body["payload"]
        assert payload["image"] == {
            "media_type": "image/jpeg",
            "width": 1280,
            "height": 720,
            "encoding": "base64",
            "data": base64.b64encode(jpeg).decode("ascii"),
        }
        assert (
            base64.b64decode(payload["image"]["data"], validate=True) == jpeg
        )
        assert payload["detections"][0] == {
            "label": "No-Helmet",
            "confidence": 0.94,
            "box": {
                "x_min": 10.0,
                "y_min": 20.0,
                "x_max": 100.0,
                "y_max": 200.0,
            },
            "track_id": None,
        }
        # The fake receives the pre-serialization dict; make sure a real JSON
        # encoder can consume the exact body without any remaining bytes.
        json.dumps(body)

    asyncio.run(scenario())


def test_transient_detection_source_frame_is_never_serialized() -> None:
    async def scenario() -> None:
        raw_frame = ImageFrame(bytes(range(12)), 2, 2, "BGR8")
        result = DetectionResult(
            detections=(
                Detection(
                    "No-Helmet",
                    0.9,
                    BoundingBox(1.0, 0.0, 2.0, 2.0),
                ),
            ),
            model_id="construction-ppe@1",
            inference_ms=3.5,
            source_frame=raw_frame,
        )
        client = FakeHttpClient([FakeResponse(200)])
        sink = await _setup_sink(client, max_attempts=1)

        await sink.consume(_envelope(result), input_port="detections")

        payload = client.calls[0]["json"]["payload"]
        assert "source_frame" not in payload
        assert payload["detections"][0]["label"] == "No-Helmet"
        assert base64.b64encode(raw_frame.data).decode("ascii") not in json.dumps(
            client.calls[0]["json"]
        )

    asyncio.run(scenario())


def test_payload_without_event_id_falls_back_to_envelope_trace_id() -> None:
    async def scenario() -> None:
        client = FakeHttpClient([FakeResponse(204)])
        sink = await _setup_sink(client, max_attempts=1)

        await sink.consume(_envelope({"status": "ok"}))

        assert client.calls[0]["headers"] == {"Idempotency-Key": "trace-fallback-456"}

    asyncio.run(scenario())


@pytest.mark.parametrize(
    "outcomes",
    [
        [FakeResponse(503), FakeResponse(503), FakeResponse(200)],
        [OSError("offline"), TimeoutError("timed out"), FakeResponse(200)],
    ],
    ids=("retryable-status", "network-errors"),
)
def test_retryable_failures_retry_to_max_attempts_with_an_unchanged_body(
    outcomes: list[FakeResponse | BaseException],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    async def scenario() -> None:
        encode_calls = 0
        original_b64encode = http_json.base64.b64encode

        def recording_b64encode(value: bytes) -> bytes:
            nonlocal encode_calls
            encode_calls += 1
            return original_b64encode(value)

        monkeypatch.setattr(http_json.base64, "b64encode", recording_b64encode)
        client = FakeHttpClient(outcomes)
        sink = await _setup_sink(
            client,
            grpc_endpoint="cmvr_es",
            max_attempts=3,
            retry_initial_s=0,
            retry_max_s=0,
        )
        envelope = _envelope(
            _alert(image=EncodedImage(b"jpeg", "image/jpeg", 2, 2))
        )

        await sink.consume(envelope)

        assert len(client.calls) == 3
        first_body = client.calls[0]["json"]
        assert all(call["json"] == first_body for call in client.calls)
        assert len({call["body_id"] for call in client.calls}) == 1
        assert first_body["grpc_ip"] == "192.168.0.119"
        assert all(
            call["headers"] == {"Idempotency-Key": "ppe-event-123"}
            for call in client.calls
        )
        assert encode_calls == 1

    asyncio.run(scenario())


def test_final_retryable_non_success_response_is_raised() -> None:
    async def scenario() -> None:
        client = FakeHttpClient([FakeResponse(503), FakeResponse(503)])
        sink = await _setup_sink(
            client,
            max_attempts=2,
            retry_initial_s=0,
            retry_max_s=0,
        )

        with pytest.raises(FakeHttpStatusError, match="503"):
            await sink.consume(_envelope(_alert()))

        assert len(client.calls) == 2

    asyncio.run(scenario())


@pytest.mark.parametrize(
    "outcome",
    [FakeResponse(400), ValueError("invalid request body")],
    ids=("non-retryable-status", "non-retryable-exception"),
)
def test_non_retryable_failure_is_raised_without_retry(
    outcome: FakeResponse | BaseException,
) -> None:
    async def scenario() -> None:
        client = FakeHttpClient([outcome, FakeResponse(200)])
        sink = await _setup_sink(
            client,
            max_attempts=3,
            retry_initial_s=0,
            retry_max_s=0,
        )

        with pytest.raises((FakeHttpStatusError, ValueError)):
            await sink.consume(_envelope(_alert()))

        assert len(client.calls) == 1

    asyncio.run(scenario())


@pytest.mark.parametrize(
    ("outcomes", "expected_attempts", "expected_status"),
    [
        ([OSError("offline"), TimeoutError("timed out")], 2, "none"),
        ([FakeResponse(503), FakeResponse(503)], 2, "503"),
        ([FakeResponse(400)], 1, "400"),
    ],
    ids=(
        "network-errors",
        "retryable-status",
        "non-retryable-status",
    ),
)
def test_log_and_drop_warns_without_failing_the_sink(
    outcomes: list[FakeResponse | BaseException],
    expected_attempts: int,
    expected_status: str,
    caplog: pytest.LogCaptureFixture,
) -> None:
    async def scenario() -> None:
        client = FakeHttpClient(outcomes)
        sink = await _setup_sink(
            client,
            failure_mode="log_and_drop",
            max_attempts=2,
            retry_initial_s=0,
            retry_max_s=0,
        )

        await sink.consume(_envelope(_alert()), input_port="alerts")

        assert len(client.calls) == expected_attempts

    with caplog.at_level(logging.WARNING, logger=http_json.__name__):
        asyncio.run(scenario())

    dropped = [
        record.getMessage()
        for record in caplog.records
        if "HTTP report dropped" in record.getMessage()
    ]
    assert len(dropped) == 1
    assert "failure_mode=log_and_drop" in dropped[0]
    assert "action=drop" in dropped[0]
    assert f"attempts={expected_attempts}" in dropped[0]
    assert f"status={expected_status}" in dropped[0]
    assert "trace_id=trace-fallback-456" in dropped[0]


def test_log_and_drop_never_swallows_cancellation() -> None:
    async def scenario() -> None:
        client = FakeHttpClient([asyncio.CancelledError()])
        sink = await _setup_sink(
            client,
            failure_mode="log_and_drop",
            max_attempts=1,
        )

        with pytest.raises(asyncio.CancelledError):
            await sink.consume(_envelope(_alert()))

    asyncio.run(scenario())


def test_log_and_drop_continues_with_the_next_envelope() -> None:
    async def scenario() -> None:
        client = FakeHttpClient([OSError("offline"), FakeResponse(204)])
        sink = await _setup_sink(
            client,
            failure_mode="log_and_drop",
            max_attempts=1,
        )

        await sink.consume(_envelope(_alert("dropped-event")))
        await sink.consume(_envelope(_alert("delivered-event")))

        assert len(client.calls) == 2
        assert client.calls[0]["headers"] == {
            "Idempotency-Key": "dropped-event"
        }
        assert client.calls[1]["headers"] == {
            "Idempotency-Key": "delivered-event"
        }

    asyncio.run(scenario())


def test_log_and_drop_does_not_swallow_an_unknown_client_error() -> None:
    async def scenario() -> None:
        client = FakeHttpClient([ValueError("client programming error")])
        sink = await _setup_sink(
            client,
            failure_mode="log_and_drop",
            max_attempts=3,
        )

        with pytest.raises(ValueError, match="client programming error"):
            await sink.consume(_envelope(_alert()))

        assert len(client.calls) == 1

    asyncio.run(scenario())


def test_log_and_drop_success_does_not_emit_a_dropped_warning(
    caplog: pytest.LogCaptureFixture,
) -> None:
    async def scenario() -> None:
        client = FakeHttpClient([FakeResponse(204)])
        sink = await _setup_sink(
            client,
            failure_mode="log_and_drop",
            max_attempts=1,
        )
        await sink.consume(_envelope(_alert()))

    with caplog.at_level(logging.WARNING, logger=http_json.__name__):
        asyncio.run(scenario())

    assert not any(
        "HTTP report dropped" in record.getMessage() for record in caplog.records
    )


@pytest.mark.parametrize(
    ("params", "message"),
    [
        ({"max_attempts": 0}, "max_attempts must be a positive integer"),
        ({"max_attempts": True}, "max_attempts must be a positive integer"),
        ({"retry_initial_s": -0.1}, "retry_initial_s must be a non-negative"),
        ({"retry_max_s": False}, "retry_max_s must be a non-negative"),
        (
            {"retry_initial_s": 2.0, "retry_max_s": 1.0},
            "retry_max_s must be >= retry_initial_s",
        ),
        ({"retry_statuses": "503"}, "retry_statuses must be a sequence"),
        ({"retry_statuses": [99]}, "must be between 100 and 599"),
        ({"retry_statuses": [600]}, "must be between 100 and 599"),
        ({"retry_statuses": ["not-a-status"]}, "must be a sequence"),
        ({"failure_mode": "warn"}, "must be 'raise' or 'log_and_drop'"),
        ({"failure_mode": ""}, "must be 'raise' or 'log_and_drop'"),
        ({"failure_mode": None}, "must be 'raise' or 'log_and_drop'"),
        ({"failure_mode": True}, "must be 'raise' or 'log_and_drop'"),
        ({"failure_mode": []}, "must be 'raise' or 'log_and_drop'"),
        ({"grpc_endpoint": ""}, "grpc_endpoint must be a non-empty"),
        ({"grpc_endpoint": "   "}, "grpc_endpoint must be a non-empty"),
        ({"grpc_endpoint": 42}, "grpc_endpoint must be a non-empty"),
    ],
)
def test_http_sink_configuration_is_validated(
    params: dict[str, Any], message: str
) -> None:
    with pytest.raises(ValueError, match=message):
        PlatformHttpJsonSink("platform", {"endpoint": "platform", **params})