"""Tests for FRTemporalAnalyzer.""" from __future__ import annotations import asyncio import sys from pathlib import Path from unittest.mock import AsyncMock import pytest # ── path setup ────────────────────────────────────────────────────────────── ROOT = Path(__file__).resolve().parents[2] SRC = ROOT if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) from services.fr_analyzer import FRTemporalAnalyzer # noqa: E402 # ── helpers ────────────────────────────────────────────────────────────────── def _det(identity: str = "Alice", confidence: float = 0.95) -> dict: """Build a single detection dict.""" return {"identity": identity, "confidence": confidence} # ── tests ──────────────────────────────────────────────────────────────────── class TestFRTemporalAnalyzer: """Tests for consistency detection logic.""" @pytest.mark.asyncio async def test_no_trigger_below_threshold(self) -> None: """Feeding detections for less than threshold_s must NOT trigger.""" callback = AsyncMock() analyzer = FRTemporalAnalyzer(consistency_threshold_s=5.0, window_s=10.0) analyzer.on_consistent_detection(callback) # Feed 3 frames over 3 seconds — below the 5s threshold. for i in range(4): await analyzer.feed([_det("Alice")], timestamp=100.0 + i) callback.assert_not_called() @pytest.mark.asyncio async def test_trigger_at_threshold(self) -> None: """Consistent detections for >= threshold_s MUST trigger the callback.""" callback = AsyncMock() analyzer = FRTemporalAnalyzer(consistency_threshold_s=5.0, window_s=10.0) analyzer.on_consistent_detection(callback) # Feed frames every 1 second for 6 seconds (0..6). for i in range(7): await analyzer.feed([_det("Alice", 0.90)], timestamp=100.0 + i) callback.assert_called_once() identity, confidence, duration = callback.call_args[0] assert identity == "Alice" assert confidence == pytest.approx(0.90, abs=0.01) assert duration >= 5.0 @pytest.mark.asyncio async def test_no_re_notification(self) -> None: """Same identity must NOT be re-notified after the first trigger.""" callback = AsyncMock() analyzer = FRTemporalAnalyzer(consistency_threshold_s=3.0, window_s=10.0) analyzer.on_consistent_detection(callback) # Trigger first notification. for i in range(5): await analyzer.feed([_det("Bob")], timestamp=100.0 + i) assert callback.call_count == 1 # Keep feeding — should NOT trigger again. for i in range(5, 10): await analyzer.feed([_det("Bob")], timestamp=100.0 + i) assert callback.call_count == 1 @pytest.mark.asyncio async def test_re_notification_after_clear_identity(self) -> None: """After clear_identity, the same person can trigger again.""" callback = AsyncMock() analyzer = FRTemporalAnalyzer(consistency_threshold_s=3.0, window_s=10.0) analyzer.on_consistent_detection(callback) # Trigger first notification. for i in range(5): await analyzer.feed([_det("Carol")], timestamp=100.0 + i) assert callback.call_count == 1 # Clear and feed more. analyzer.clear_identity("Carol") for i in range(5, 10): await analyzer.feed([_det("Carol")], timestamp=100.0 + i) assert callback.call_count == 2 @pytest.mark.asyncio async def test_reset_clears_everything(self) -> None: """reset() must clear history and notified identities.""" callback = AsyncMock() analyzer = FRTemporalAnalyzer(consistency_threshold_s=3.0, window_s=10.0) analyzer.on_consistent_detection(callback) # Build up some history. for i in range(5): await analyzer.feed([_det("Dave")], timestamp=100.0 + i) assert callback.call_count == 1 analyzer.reset() # After reset, need to build up from scratch again. for i in range(6): await analyzer.feed([_det("Dave")], timestamp=200.0 + i) assert callback.call_count == 2 @pytest.mark.asyncio async def test_presence_ratio_requirement(self) -> None: """An identity must appear in >= 60% of frames to be considered consistent.""" callback = AsyncMock() analyzer = FRTemporalAnalyzer(consistency_threshold_s=5.0, window_s=10.0) analyzer.on_consistent_detection(callback) # Feed 10 frames over 10 seconds. "Eve" appears sporadically in # only 3 out of 10 frames (30% < 60%), interleaved with "Frank". for i in range(10): if i in (0, 4, 8): dets = [_det("Eve")] else: dets = [_det("Frank")] await analyzer.feed(dets, timestamp=100.0 + i) # Eve should NOT have triggered (only 30% presence ratio). eve_calls = [ c for c in callback.call_args_list if c[0][0] == "Eve" ] assert len(eve_calls) == 0 # Frank appears in 7 out of 10 frames (70% >= 60%) and spans >=5s. frank_calls = [ c for c in callback.call_args_list if c[0][0] == "Frank" ] assert len(frank_calls) == 1 @pytest.mark.asyncio async def test_multiple_identities(self) -> None: """Two different identities both consistent should both trigger.""" callback = AsyncMock() analyzer = FRTemporalAnalyzer(consistency_threshold_s=3.0, window_s=10.0) analyzer.on_consistent_detection(callback) # Both Alice and Bob present in every frame. for i in range(5): await analyzer.feed( [_det("Alice", 0.85), _det("Bob", 0.70)], timestamp=100.0 + i, ) triggered_identities = {c[0][0] for c in callback.call_args_list} assert "Alice" in triggered_identities assert "Bob" in triggered_identities @pytest.mark.asyncio async def test_unknown_identity_defaults(self) -> None: """Detections without an 'identity' key should default to 'unknown'.""" callback = AsyncMock() analyzer = FRTemporalAnalyzer(consistency_threshold_s=3.0, window_s=10.0) analyzer.on_consistent_detection(callback) for i in range(5): await analyzer.feed( [{"confidence": 0.5}], # no "identity" key timestamp=100.0 + i, ) callback.assert_called_once() assert callback.call_args[0][0] == "unknown" @pytest.mark.asyncio async def test_no_callback_registered(self) -> None: """Feed should not error when no callback is registered.""" analyzer = FRTemporalAnalyzer(consistency_threshold_s=3.0, window_s=10.0) # Should not raise. for i in range(5): await analyzer.feed([_det("Alice")], timestamp=100.0 + i) @pytest.mark.asyncio async def test_min_frames_requirement(self) -> None: """Fewer than MIN_FRAMES in the window must not trigger.""" callback = AsyncMock() # Use a very short threshold but long time gaps so only 2 frames # fall inside the threshold window. analyzer = FRTemporalAnalyzer(consistency_threshold_s=2.0, window_s=10.0) analyzer.on_consistent_detection(callback) # Two frames 3s apart — the second frame's threshold window (t-2..t) # only contains the second frame itself. await analyzer.feed([_det("Alice")], timestamp=100.0) await analyzer.feed([_det("Alice")], timestamp=103.0) callback.assert_not_called() @pytest.mark.asyncio async def test_sliding_window_prunes_old_entries(self) -> None: """Entries older than window_s must be pruned.""" analyzer = FRTemporalAnalyzer(consistency_threshold_s=3.0, window_s=5.0) # Feed at t=100..104 (5 frames in window). for i in range(5): await analyzer.feed([_det("Alice")], timestamp=100.0 + i) # Feed at t=110 — entries at 100..104 are outside the 5s window. await analyzer.feed([_det("Alice")], timestamp=110.0) # Internal history should only contain the t=110 entry. assert len(analyzer._detection_history) == 1 assert analyzer._detection_history[0][0] == 110.0