detections/e2e/tests/test_resilience.py

import json
import os
import threading
import time
import uuid

import pytest
import requests

_DETECT_TIMEOUT = 60
_MEDIA = os.environ.get("MEDIA_DIR", "/media")


def _ai_config_video() -> dict:
    return {
        "probability_threshold": 0.25,
        "tracking_intersection_threshold": 0.6,
        "altitude": 400,
        "focal_length": 24,
        "sensor_width": 23.5,
        "paths": [f"{_MEDIA}/video_short01.mp4"],
        "frame_period_recognition": 4,
        "frame_recognition_seconds": 2,
    }


def test_ft_n_06_loader_unreachable_during_init_health(
    http_client, mock_loader_url, image_small
):
    h0 = http_client.get("/health")
    h0.raise_for_status()
    if h0.json().get("aiAvailability") != "None":
        pytest.skip("engine already warm")
    requests.post(
        f"{mock_loader_url}/mock/config", json={"mode": "error"}, timeout=10
    ).raise_for_status()
    files = {"file": ("n06.jpg", image_small, "image/jpeg")}
    r = http_client.post("/detect", files=files, timeout=_DETECT_TIMEOUT)
    assert r.status_code != 500
    h = http_client.get("/health")
    assert h.status_code == 200
    d = h.json()
    assert d["status"] == "healthy"
    assert d.get("errorMessage") is None


@pytest.mark.slow
@pytest.mark.timeout(120)
def test_ft_n_07_annotations_unreachable_detection_continues(
    warm_engine,
    http_client,
    jwt_token,
    mock_annotations_url,
    sse_client_factory,
):
    requests.post(
        f"{mock_annotations_url}/mock/config", json={"mode": "error"}, timeout=10
    ).raise_for_status()
    media_id = f"res-n07-{uuid.uuid4().hex}"
    body = _ai_config_video()
    headers = {"Authorization": f"Bearer {jwt_token}"}
    collected = []
    thread_exc = []
    done = threading.Event()

    def _listen():
        try:
            with sse_client_factory() as sse:
                time.sleep(0.3)
                for event in sse.events():
                    if not event.data or not str(event.data).strip():
                        continue
                    data = json.loads(event.data)
                    if data.get("mediaId") != media_id:
                        continue
                    collected.append(data)
                    if (
                        data.get("mediaStatus") == "AIProcessed"
                        and data.get("mediaPercent") == 100
                    ):
                        break
        except BaseException as e:
            thread_exc.append(e)
        finally:
            done.set()

    th = threading.Thread(target=_listen, daemon=True)
    th.start()
    time.sleep(0.5)
    pr = http_client.post(f"/detect/{media_id}", json=body, headers=headers)
    assert pr.status_code == 200
    ok = done.wait(timeout=120)
    assert ok
    th.join(timeout=5)
    assert not thread_exc
    assert any(
        e.get("mediaStatus") == "AIProcessed" and e.get("mediaPercent") == 100
        for e in collected
    )


def test_nft_res_01_loader_outage_after_init(
    warm_engine, http_client, mock_loader_url, image_small
):
    requests.post(
        f"{mock_loader_url}/mock/config", json={"mode": "error"}, timeout=10
    ).raise_for_status()
    files = {"file": ("r1.jpg", image_small, "image/jpeg")}
    r = http_client.post("/detect", files=files, timeout=_DETECT_TIMEOUT)
    assert r.status_code == 200
    assert isinstance(r.json(), list)
    h = http_client.get("/health")
    assert h.status_code == 200
    hd = h.json()
    assert hd["status"] == "healthy"
    assert hd.get("errorMessage") is None


@pytest.mark.slow
@pytest.mark.timeout(120)
def test_nft_res_02_annotations_outage_during_async_detection(
    warm_engine,
    http_client,
    jwt_token,
    mock_annotations_url,
    sse_client_factory,
):
    media_id = f"res-n02-{uuid.uuid4().hex}"
    body = _ai_config_video()
    headers = {"Authorization": f"Bearer {jwt_token}"}
    collected = []
    thread_exc = []
    done = threading.Event()

    def _listen():
        try:
            with sse_client_factory() as sse:
                time.sleep(0.3)
                for event in sse.events():
                    if not event.data or not str(event.data).strip():
                        continue
                    data = json.loads(event.data)
                    if data.get("mediaId") != media_id:
                        continue
                    collected.append(data)
                    if (
                        data.get("mediaStatus") == "AIProcessed"
                        and data.get("mediaPercent") == 100
                    ):
                        break
        except BaseException as e:
            thread_exc.append(e)
        finally:
            done.set()

    th = threading.Thread(target=_listen, daemon=True)
    th.start()
    time.sleep(0.5)
    pr = http_client.post(f"/detect/{media_id}", json=body, headers=headers)
    assert pr.status_code == 200
    requests.post(
        f"{mock_annotations_url}/mock/config", json={"mode": "error"}, timeout=10
    ).raise_for_status()
    ok = done.wait(timeout=120)
    assert ok
    th.join(timeout=5)
    assert not thread_exc
    assert any(
        e.get("mediaStatus") == "AIProcessed" and e.get("mediaPercent") == 100
        for e in collected
    )


def test_nft_res_03_transient_loader_first_fail(
    mock_loader_url, http_client, image_small
):
    requests.post(
        f"{mock_loader_url}/mock/config", json={"mode": "first_fail"}, timeout=10
    ).raise_for_status()
    files = {"file": ("r3a.jpg", image_small, "image/jpeg")}
    r1 = http_client.post("/detect", files=files, timeout=_DETECT_TIMEOUT)
    files2 = {"file": ("r3b.jpg", image_small, "image/jpeg")}
    r2 = http_client.post("/detect", files=files2, timeout=_DETECT_TIMEOUT)
    assert r2.status_code == 200
    if r1.status_code != 200:
        assert r1.status_code != 500


@pytest.mark.skip(
    reason="Requires docker compose restart capability not available in e2e-runner"
)
def test_nft_res_04_service_restart():
    pass