Spaces:
Build error
Build error
| """Tests for SignalBuffer.""" | |
| from __future__ import annotations | |
| import asyncio | |
| import time | |
| import pytest | |
| from memorybridge.core.schemas import ( | |
| AffectSignal, | |
| AirSignChar, | |
| GestureSignal, | |
| SignalBufferState, | |
| VisionSnapshot, | |
| ) | |
| from memorybridge.vision_path.signal_buffer import SignalBuffer | |
| def make_snapshot( | |
| gesture_class="neutral", | |
| gesture_confidence=1.0, | |
| affect_class="neutral", | |
| affect_confidence=0.9, | |
| air_sign_char=None, | |
| ) -> VisionSnapshot: | |
| ts = time.time() | |
| return VisionSnapshot( | |
| timestamp=ts, | |
| gestures=[GestureSignal(gesture_class=gesture_class, confidence=gesture_confidence, timestamp=ts)], | |
| affect=AffectSignal(affect_class=affect_class, confidence=affect_confidence, timestamp=ts), | |
| air_sign_char=AirSignChar(character=air_sign_char, confidence=0.9, timestamp=ts) if air_sign_char else None, | |
| ) | |
| class TestSignalBuffer: | |
| async def test_push_and_size(self): | |
| buf = SignalBuffer(buffer_size=4) | |
| await buf.push(make_snapshot()) | |
| assert len(buf._buffer) == 1 | |
| async def test_max_size_rolling(self): | |
| buf = SignalBuffer(buffer_size=3) | |
| for _ in range(5): | |
| await buf.push(make_snapshot()) | |
| assert len(buf._buffer) == 3 # maxlen enforced | |
| async def test_get_state_returns_signal_buffer_state(self): | |
| """get_state() returns a SignalBufferState with the pushed snapshot.""" | |
| buf = SignalBuffer(buffer_size=4) | |
| await buf.push(make_snapshot()) | |
| state = await buf.get_state() | |
| assert isinstance(state, SignalBufferState) | |
| assert len(state.snapshots) == 1 | |
| def test_clear(self): | |
| buf = SignalBuffer(buffer_size=4) | |
| buf._buffer.append(make_snapshot()) | |
| buf.clear() | |
| assert len(buf._buffer) == 0 | |
| def test_aggregate_gesture_neutral_returns_none(self): | |
| """Neutral gestures are skipped — all-neutral snapshots return None.""" | |
| buf = SignalBuffer() | |
| result = buf._aggregate_gesture([make_snapshot(gesture_class="neutral")]) | |
| assert result is None | |
| def test_aggregate_gesture_non_neutral_returned(self): | |
| """Non-neutral gesture with highest confidence is returned.""" | |
| buf = SignalBuffer() | |
| snap = make_snapshot(gesture_class="thumbs_up", gesture_confidence=0.95) | |
| result = buf._aggregate_gesture([snap]) | |
| assert result is not None | |
| assert result.gesture_class == "thumbs_up" | |
| def test_aggregate_affect_empty_snapshots_returns_none(self): | |
| """Empty snapshot list → None.""" | |
| buf = SignalBuffer() | |
| result = buf._aggregate_affect([]) | |
| assert result is None | |
| def test_aggregate_affect_non_neutral_dominant(self): | |
| """Happy affect appearing in ≥ half the snapshots returns AffectSignal.""" | |
| buf = SignalBuffer() | |
| snaps = [make_snapshot(affect_class="happy", affect_confidence=0.9) for _ in range(3)] | |
| result = buf._aggregate_affect(snaps) | |
| assert result is not None | |
| assert result.affect_class == "happy" | |
| def test_confirm_air_sign_insufficient_windows_returns_empty(self): | |
| """Single snapshot with air_sign_char — insufficient windows → [].""" | |
| buf = SignalBuffer() | |
| result = buf._confirm_air_sign_sequence([make_snapshot(air_sign_char="B")]) | |
| # confirmation_windows defaults to 2; one snapshot is not enough | |
| assert isinstance(result, list) | |
| assert len(result) == 0 | |