Behavision/tests/test_detector_concurrency.py

"""cv2.FaceDetectorYN caches its input size and is not thread-safe, so camera
workers must not share one. Skipped when the model is absent, matching the
faiss-optional pattern in test_index.py — the suite stays runnable with no
models installed.

The premise is proved in a SUBPROCESS, and that is the whole lesson of this
file. The first version raced a shared detector in-process and asserted that
an exception came back, because on OpenCV 4.11 one usually did. It is a data
race in C++: what it produces is undefined, and on 4.14 what it mostly
produces is a segmentation fault. Measured, nine runs each, same machine:

    numpy 1.26 / cv2 4.11    9 passed, 0 crashed
    numpy 2.0  / cv2 4.11    8 passed, 1 crashed
    numpy 1.26 / cv2 4.14    3 passed, 6 crashed
    numpy 2.0  / cv2 4.14    2 passed, 7 crashed

numpy is not the variable; OpenCV is, and 4.11 crashing once says the hazard
was always there and 4.11 merely survived it. A test that takes the whole
suite down two runs in three is worse than no test: it turns "we upgraded
OpenCV" into a CI failure with no failing assertion in it, which is the
hardest kind to read.

None of this reaches the product. `Engine._build_worker` constructs a
FaceDetector per camera, which is what the rule says and what the second test
here guards.
"""
import subprocess
import sys
import textwrap
import threading
from pathlib import Path

import numpy as np
import pytest

from behavision.config import Config
from behavision.detection import YUNET_FILENAME, FaceDetector

ROOT = Path(__file__).resolve().parent.parent
MODELS = ROOT / "models"
pytestmark = pytest.mark.skipif(not (MODELS / YUNET_FILENAME).exists(),
                                reason="YuNet model not installed")

# A shared detector fails probabilistically, so one clean attempt proves
# nothing. Several do: the premise holds if ANY attempt misbehaves, and only
# an unbroken run of clean ones is evidence it has stopped being true.
ATTEMPTS = 6


def _detector():
    d = Config().detection
    return FaceDetector(MODELS, d.score_threshold, d.nms_threshold,
                        d.max_faces, d.min_face_px)


def _hammer(det, size, errors, n=40):
    w, h = size
    frame = np.zeros((h, w, 3), dtype=np.uint8)
    for _ in range(n):
        try:
            det.detect(frame)
        except Exception as exc:  # noqa: BLE001 - cv2 raises on size mismatch
            errors.append(exc)
            return


def _race(det_a, det_b):
    errors = []
    threads = [threading.Thread(target=_hammer, args=(det_a, (1280, 720), errors)),
               threading.Thread(target=_hammer, args=(det_b, (640, 480), errors))]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return errors


_SHARED_RACE = textwrap.dedent("""
    import sys
    sys.path.insert(0, {root!r})
    from tests.test_detector_concurrency import _detector, _race
    shared = _detector()
    print("raced" if _race(shared, shared) else "clean")
""")


def _shared_race_outcome():
    """Run one shared-detector race out of process.

    Returns "raced" (an exception came back), "crashed" (the process died,
    which is the same premise arriving by a blunter route) or "clean".
    """
    proc = subprocess.run([sys.executable, "-c", _SHARED_RACE.format(root=str(ROOT))],
                          capture_output=True, text=True, timeout=120)
    if proc.returncode != 0:
        return "crashed"
    return proc.stdout.strip().splitlines()[-1] if proc.stdout.strip() else "clean"


def test_a_shared_detector_really_does_race():
    """Guards the premise: if this ever stops failing, the test below is
    proving nothing and the per-camera split can be revisited."""
    seen = [_shared_race_outcome() for _ in range(ATTEMPTS)]
    assert any(o != "clean" for o in seen), (
        f"a shared detector survived {ATTEMPTS} races ({seen}) - if that is "
        "reproducible, cv2.FaceDetectorYN may have become thread-safe and the "
        "per-camera rule in CLAUDE.md can be revisited"
    )


def test_per_camera_detectors_do_not_race():
    assert _race(_detector(), _detector()) == []