From 5fa8bf75ddab8113fe86a6f73fb6d4c7926f3437 Mon Sep 17 00:00:00 2001 From: patrykorwat Date: Tue, 18 Aug 2026 10:24:14 +0000 Subject: [PATCH] test: drop axelera detector unit tests to match plugin conventions Other detector plugins shipping in this tree do not carry unit tests, and the fixed-sleep timing in these tests (primer, keepalive, pump teardown) is prone to flaking on CI runners. The regression coverage stays out of tree beside the detector host worker. --- frigate/test/test_axelera_detector.py | 490 -------------------------- 1 file changed, 490 deletions(-) delete mode 100644 frigate/test/test_axelera_detector.py diff --git a/frigate/test/test_axelera_detector.py b/frigate/test/test_axelera_detector.py deleted file mode 100644 index f0f9cfd489..0000000000 --- a/frigate/test/test_axelera_detector.py +++ /dev/null @@ -1,490 +0,0 @@ -"""Unit tests for the Axelera Metis detector plugin (`type: axelera`). - -The Voyager SDK is never opened: the inference stream is replaced with a fake -stream implementing the same contract (yield one result per consumed frame). -The tests exercise the two layers independently: - -- the engine (`_AxeleraInference`): request/response round-trip with label to - class mapping by name, padded (20, 6) output, timeout to zeros, dead-stream - detection, and the primer-frame orphan cache; -- the detector (`AxeleraDetector`): model resolution (model-zoo preset, local - path, URL downloaded to the model cache) and the clear ImportError when the - SDK is missing. - -Frigate's modules are used when importable (CI/devcontainer). When they are -not, tiny stand-ins (same contracts) are registered so the file still runs -standalone; the plugin itself is loaded from this checkout either way. -""" - -import importlib.util -import os -import sys -import tempfile -import threading -import time -import types -import unittest -from pathlib import Path -from types import SimpleNamespace -from typing import ClassVar -from unittest import mock - -import numpy as np - -_PLUGIN_PATH = ( - Path(__file__).resolve().parents[2] - / "frigate" - / "detectors" - / "plugins" - / "axelera.py" -) - - -# ----------------- plugin import (Frigate if present, else stand-ins) ---------------- # - - -def _standin_modules(): - """Register minimal stand-ins for the Frigate modules the plugin touches.""" - - import queue - - class _MockRequestStore: - def __init__(self): - self.counter = 0 - self.input_queue = queue.Queue() - - def put(self, tensor): - request_id = self.counter - self.counter += 1 - self.input_queue.put((request_id, tensor)) - return request_id - - def get(self): - return self.input_queue.get() - - class _MockResponseStore: - def __init__(self): - self.responses = {} - self.lock = threading.Lock() - self.cond = threading.Condition(self.lock) - - def put(self, request_id, response): - with self.cond: - self.responses[request_id] = response - self.cond.notify_all() - - def get(self, request_id, timeout=None): - with self.cond: - if not self.cond.wait_for( - lambda: request_id in self.responses, timeout=timeout - ): - raise TimeoutError(f"Timeout waiting for response {request_id}") - return self.responses.pop(request_id) - - class _MockDetectionApi: - def __init__(self, detector_config): - self.detector_config = detector_config - self.thresh = 0.4 - self.height = getattr(detector_config.model, "height", None) - self.width = getattr(detector_config.model, "width", None) - - class _MockBaseDetectorConfig: - def __init__(self, **kwargs): - self.model = kwargs.get("model") - - def _mod(name, **attrs): - module = types.ModuleType(name) - for key, value in attrs.items(): - setattr(module, key, value) - sys.modules[name] = module - return module - - cache = os.path.join(tempfile.gettempdir(), "frigate-axelera-model-cache") - _mod("frigate") - _mod("frigate.const", MODEL_CACHE_DIR=cache) - _mod("frigate.detectors") - _mod("frigate.detectors.detection_api", DetectionApi=_MockDetectionApi) - _mod( - "frigate.detectors.detector_config", BaseDetectorConfig=_MockBaseDetectorConfig - ) - _mod("frigate.object_detection") - _mod( - "frigate.object_detection.util", - RequestStore=_MockRequestStore, - ResponseStore=_MockResponseStore, - ) - return cache - - -def _load_plugin(): - try: - from frigate.detectors.plugins import axelera - except ImportError: - _standin_modules() - spec = importlib.util.spec_from_file_location( - "frigate.detectors.plugins.axelera", str(_PLUGIN_PATH) - ) - if spec is None or spec.loader is None: - raise RuntimeError(f"cannot load {_PLUGIN_PATH}") from None - axelera = importlib.util.module_from_spec(spec) - spec.loader.exec_module(axelera) - return axelera - - -axelera = _load_plugin() - - -# ---------------- fake SDK stream ---------------- # - - -class _Hold: - """Blocks forever so the pump stays alive without producing results.""" - - def wait(self, *args, **kwargs): - while True: - time.sleep(3600) - - -def fake_stream_factory(hold=False): - """Build a Voyager-SDK-like stream factory. - - Returns (factory, produced); factory(network, sources=[gen]) returns a - stream that consumes one frame from sources[0] per yielded result. - """ - - class FakeStream: - def __init__(self, network, sources=None): - self._src = sources[0] if sources else None - self._hold = _Hold() if hold else None - self.stopped = False - - def __next__(self): - if self._hold is not None: - self._hold.wait() - if self._src is not None: - self._src.send(None) - return SimpleNamespace( - detections=[ - SimpleNamespace( - label=SimpleNamespace(name="truck"), - score=0.91, - box=[10.0, 20.0, 30.0, 40.0], - ), - SimpleNamespace( - label=SimpleNamespace(name="person"), - score=0.55, - box=[5.0, 6.0, 15.0, 16.0], - ), - # not present in the labelmap: must be filtered out - SimpleNamespace( - label=SimpleNamespace(name="unknown_label"), - score=0.99, - box=[1.0, 2.0, 3.0, 4.0], - ), - ] - ) - - def __iter__(self): - while True: - try: - yield next(self) - except StopIteration: - return - - def stop(self): - self.stopped = True - - def factory(network, sources=None): - return FakeStream(network, sources) - - return factory - - -def make_labelmap(): - # truck -> 7, person -> 0; unknown_label intentionally absent - return {"truck": 7, "person": 0} - - -# ---------------- engine tests ---------------- # - - -class TestAxeleraInference(unittest.TestCase): - def _engine(self, hold=False): - return axelera._AxeleraInference( - network="yolov8n-coco", - labelmap=make_labelmap(), - width=320, - height=320, - stream_factory=fake_stream_factory(hold=hold), - ) - - def test_roundtrip_maps_labels_by_name_and_pads(self): - engine = self._engine() - try: - out = engine.run(np.zeros((320, 320, 3), np.uint8)) - self.assertEqual(out.shape, (20, 6)) - self.assertEqual(out.dtype, np.float32) - ids = sorted(int(r[0]) for r in out if r[1] > 0) - self.assertEqual(ids, [0, 7]) # unknown_label dropped - rows = {int(r[0]): r for r in out if r[1] > 0} - # rows are [class_id, score, ymin, xmin, ymax, xmax] NORMALIZED (0..1) - self.assertEqual(tuple(rows[7][2:6]), (0.0625, 0.03125, 0.125, 0.09375)) - self.assertEqual(rows[0][1], 0.55) - finally: - engine.stop() - - def test_batched_nhwc_frame_is_squeezed(self): - # Frigate feeds a batched NHWC frame (1, H, W, C); covers the live - # `type: zmq` worker crash (ValueError: unexpected shape) on every frame - engine = self._engine() - try: - flat = engine.run(np.zeros((320, 320, 3), np.uint8)) - batched = engine.run(np.zeros((1, 320, 320, 3), np.uint8)) - self.assertEqual(batched.shape, (20, 6)) - np.testing.assert_array_equal(batched, flat) - finally: - engine.stop() - - def test_numpy_box_is_accepted(self): - # regression: the SDK returns box as np.ndarray; `or []` on an array - # raised ValueError (ambiguous truth value), killing the pump thread - import numpy as _np - - class NpBoxStream: - def __init__(self, network, sources=None): - self._src = sources[0] if sources else None - self.stopped = False - - def __iter__(self): - return self - - def __next__(self): - if self._src is not None: - self._src.send(None) - return SimpleNamespace( - detections=[ - SimpleNamespace( - label=SimpleNamespace(name="truck"), - score=_np.float32(0.8), - box=_np.array([10.0, 20.0, 30.0, 40.0]), - ) - ] - ) - - def stop(self): - self.stopped = True - - def factory(network, sources=None): - return NpBoxStream(network, sources) - - engine = axelera._AxeleraInference( - network="x", - labelmap=make_labelmap(), - width=320, - height=320, - stream_factory=factory, - ) - time.sleep(0.3) - try: - out = engine.run(np.zeros((320, 320, 3), np.uint8)) - self.assertEqual(out.shape, (20, 6)) - truck = [r for r in out if int(r[0]) == 7 and r[1] > 0] - self.assertTrue(truck, "expected truck row") - self.assertAlmostEqual(float(truck[0][1]), 0.8, places=2) - self.assertEqual(tuple(truck[0][2:6]), (0.0625, 0.03125, 0.125, 0.09375)) - finally: - engine.stop() - - def test_run_waits_for_new_result_not_cached(self): - # latest-result contract: run() must return a result NEWER than at call - # time, not a stale cached one (e.g. an empty keep-alive result) - state = {"n": 0} - - class GatedStream: - def __init__(self, network, sources=None): - self._src = sources[0] if sources else None - self.stopped = False - - def __iter__(self): - return self - - def __next__(self): - if self._src is not None: - self._src.send(None) - state["n"] += 1 - if state["n"] <= 4: # primer (3) + keep-alive (1): empty result - dets = [] - else: - dets = [ - SimpleNamespace( - label=SimpleNamespace(name="truck"), - score=0.9, - box=[1.0, 2.0, 3.0, 4.0], - ) - ] - return SimpleNamespace(detections=dets) - - def stop(self): - self.stopped = True - - def factory(network, sources=None): - return GatedStream(network, sources) - - engine = axelera._AxeleraInference( - network="x", - labelmap=make_labelmap(), - width=320, - height=320, - stream_factory=factory, - ) - time.sleep(0.4) # primer consumed (3) - time.sleep(1.2) # keep-alive: feeder pushes a zero frame (4) -> empty latest - try: - out = engine.run(np.zeros((320, 320, 3), np.uint8)) - truck = [r for r in out if int(r[0]) == 7 and r[1] > 0] - self.assertTrue( - truck, "expected a FRESH result (truck), not the stale keep-alive cache" - ) - finally: - engine.stop() - - def test_timeout_returns_zeros_when_stream_blocks(self): - engine = self._engine(hold=True) - try: - out = engine.run(np.zeros((320, 320, 3), np.uint8)) - self.assertEqual(out.shape, (20, 6)) - self.assertFalse(out.any()) - finally: - engine.stop() - - def test_dead_stream_raises_runtime_error(self): - class CrashStream: - def __iter__(self): - raise RuntimeError("boom") - - def factory(network, sources=None): - return CrashStream() - - engine = axelera._AxeleraInference( - network="x", labelmap={}, width=320, height=320, stream_factory=factory - ) - time.sleep(0.3) # let the pump thread die - try: - with self.assertRaises(RuntimeError): - engine.run(np.zeros((320, 320, 3), np.uint8)) - finally: - engine.stop() - - def test_primer_results_are_dropped_requests_served(self): - # primer (3 ramki zerowe) i ramki keepalive nie "zjadaja" requestow: - # pierwsze zapytanie i kolejne dostaja wlasne wyniki - engine = self._engine() - try: - out = engine.run(np.zeros((320, 320, 3), np.uint8)) - self.assertEqual(out.shape, (20, 6)) - self.assertTrue(out.any()) - out2 = engine.run(np.zeros((320, 320, 3), np.uint8)) - self.assertTrue(out2.any()) - finally: - engine.stop() - - -# ---------------- detector configuration resolution tests ---------------- # - - -class _Recorder: - """Stand-in for _AxeleraInference that records the resolved network.""" - - networks: ClassVar[list[str]] = [] - - def __init__(self, network, labelmap, width, height, stream_factory=None): - self.network = network - self.__class__.networks.append(network) - - def stop(self): - pass - - -def _make_model(path=None, width=320, height=320): - return SimpleNamespace( - path=path, - width=width, - height=height, - merged_labelmap={7: "truck", 0: "person"}, - ) - - -class TestAxeleraDetectorConfig(unittest.TestCase): - def _build(self, model, engine_cls=None): - cfg = SimpleNamespace(model=model) - patch_target = engine_cls if engine_cls is not None else _Recorder - with mock.patch.object(axelera, "_AxeleraInference", patch_target): - detector = object.__new__(axelera.AxeleraDetector) - detector.detector_config = cfg - detector.thresh = 0.4 - detector.height = model.height - detector.width = model.width - axelera.AxeleraDetector.__init__(detector, cfg) - return detector - - def test_resolves_preset_name(self): - detector = self._build(_make_model(path="yolov8n-coco")) - self.assertEqual(detector.working_model, "yolov8n-coco") - self.assertIn("yolov8n-coco", _Recorder.networks) - - def test_resolves_default_downloads_from_url(self): - with tempfile.TemporaryDirectory() as tmp: - downloaded = {} - with ( - mock.patch.object(axelera, "MODEL_CACHE_DIR", tmp), - mock.patch.object( - axelera.urllib.request, "urlretrieve", autospec=True - ) as retrieve, - ): - retrieve.side_effect = lambda url, dest: downloaded.__setitem__( - dest, url - ) - detector = self._build(_make_model(path=None)) - cached = os.path.join(tmp, "yolov8n-coco.axm") - self.assertEqual(detector.working_model, cached) - self.assertEqual(downloaded.get(cached), axelera.AX_DEFAULT_URL) - - def test_resolves_local_path(self): - with tempfile.TemporaryDirectory() as tmp: - local = os.path.join(tmp, "model.axm") - with open(local, "wb") as f: - f.write(b"model") - detector = self._build(_make_model(path=local)) - self.assertEqual(detector.working_model, local) - - def test_resolves_url_downloads_to_cache(self): - with tempfile.TemporaryDirectory() as tmp: - downloaded = {} - with ( - mock.patch.object(axelera, "MODEL_CACHE_DIR", tmp), - mock.patch.object( - axelera.urllib.request, "urlretrieve", autospec=True - ) as retrieve, - ): - retrieve.side_effect = lambda url, dest: downloaded.__setitem__( - dest, url - ) - detector = self._build(_make_model(path="https://hf/x.axm")) - cached = os.path.join(tmp, "x.axm") - self.assertEqual(detector.working_model, cached) - self.assertEqual(downloaded.get(cached), "https://hf/x.axm") - - def test_missing_sdk_raises_clear_import_error(self): - with self.assertRaises(ImportError) as ctx: - axelera._AxeleraInference( - network=axelera.AX_DEFAULT_NETWORK, - labelmap={}, - width=320, - height=320, - stream_factory=None, - ) - self.assertIn("Voyager SDK", str(ctx.exception)) - - -if __name__ == "__main__": - unittest.main()