David Gwilliam
15de46722a
- Add EventBus class with pub/sub messaging (thread-safe) - Add emitter Protocol classes (EventEmitter, Startable, Stoppable) - Add event emission to NtfyPoller (NtfyMessageEvent) - Add event emission to MicMonitor (MicLevelEvent) - Update StreamController to publish stream start/end events - Add comprehensive tests for eventbus and emitters modules
150 lines
4.0 KiB
Python
150 lines
4.0 KiB
Python
"""
|
|
Tests for engine.mic module.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
|
|
from engine.events import MicLevelEvent
|
|
|
|
|
|
class TestMicMonitorImport:
|
|
"""Tests for module import behavior."""
|
|
|
|
def test_mic_monitor_imports_without_error(self):
|
|
"""MicMonitor can be imported even without sounddevice."""
|
|
from engine.mic import MicMonitor
|
|
|
|
assert MicMonitor is not None
|
|
|
|
|
|
class TestMicMonitorInit:
|
|
"""Tests for MicMonitor initialization."""
|
|
|
|
def test_init_sets_threshold(self):
|
|
"""Threshold is set correctly."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor(threshold_db=60)
|
|
assert monitor.threshold_db == 60
|
|
|
|
def test_init_defaults(self):
|
|
"""Default values are set correctly."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor()
|
|
assert monitor.threshold_db == 50
|
|
|
|
def test_init_db_starts_at_negative(self):
|
|
"""_db starts at negative value."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor()
|
|
assert monitor.db == -99.0
|
|
|
|
|
|
class TestMicMonitorProperties:
|
|
"""Tests for MicMonitor properties."""
|
|
|
|
def test_excess_returns_positive_when_above_threshold(self):
|
|
"""excess returns positive value when above threshold."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor(threshold_db=50)
|
|
with patch.object(monitor, "_db", 60.0):
|
|
assert monitor.excess == 10.0
|
|
|
|
def test_excess_returns_zero_when_below_threshold(self):
|
|
"""excess returns zero when below threshold."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor(threshold_db=50)
|
|
with patch.object(monitor, "_db", 40.0):
|
|
assert monitor.excess == 0.0
|
|
|
|
|
|
class TestMicMonitorAvailable:
|
|
"""Tests for MicMonitor.available property."""
|
|
|
|
def test_available_is_bool(self):
|
|
"""available returns a boolean."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor()
|
|
assert isinstance(monitor.available, bool)
|
|
|
|
|
|
class TestMicMonitorStop:
|
|
"""Tests for MicMonitor.stop method."""
|
|
|
|
def test_stop_does_nothing_when_no_stream(self):
|
|
"""stop() does nothing if no stream exists."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor()
|
|
monitor.stop()
|
|
assert monitor._stream is None
|
|
|
|
|
|
class TestMicMonitorEventEmission:
|
|
"""Tests for MicMonitor event emission."""
|
|
|
|
def test_subscribe_adds_callback(self):
|
|
"""subscribe() adds a callback."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor()
|
|
def callback(e):
|
|
return None
|
|
|
|
monitor.subscribe(callback)
|
|
|
|
assert callback in monitor._subscribers
|
|
|
|
def test_unsubscribe_removes_callback(self):
|
|
"""unsubscribe() removes a callback."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor()
|
|
def callback(e):
|
|
return None
|
|
monitor.subscribe(callback)
|
|
|
|
monitor.unsubscribe(callback)
|
|
|
|
assert callback not in monitor._subscribers
|
|
|
|
def test_emit_calls_subscribers(self):
|
|
"""_emit() calls all subscribers."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor()
|
|
received = []
|
|
|
|
def callback(event):
|
|
received.append(event)
|
|
|
|
monitor.subscribe(callback)
|
|
event = MicLevelEvent(
|
|
db_level=60.0, excess_above_threshold=10.0, timestamp=datetime.now()
|
|
)
|
|
monitor._emit(event)
|
|
|
|
assert len(received) == 1
|
|
assert received[0].db_level == 60.0
|
|
|
|
def test_emit_handles_subscriber_exception(self):
|
|
"""_emit() handles exceptions in subscribers gracefully."""
|
|
from engine.mic import MicMonitor
|
|
|
|
monitor = MicMonitor()
|
|
|
|
def bad_callback(event):
|
|
raise RuntimeError("test")
|
|
|
|
monitor.subscribe(bad_callback)
|
|
event = MicLevelEvent(
|
|
db_level=60.0, excess_above_threshold=10.0, timestamp=datetime.now()
|
|
)
|
|
monitor._emit(event)
|