[claude] Mumble voice bridge — Alexander ↔ Timmy co-play audio (#858) (#1324)

tests/integrations/test_mumble_bridge.py

@@ -0,0 +1,528 @@
+"""Unit tests for the Mumble voice bridge integration."""
+
+from __future__ import annotations
+
+import struct
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+def _pcm_silence(ms: int = 10, sample_rate: int = 48000) -> bytes:
+    """Return *ms* milliseconds of 16-bit 48 kHz silent PCM."""
+    n = sample_rate * ms // 1000
+    return struct.pack(f"<{n}h", *([0] * n))
+
+
+def _pcm_tone(ms: int = 10, sample_rate: int = 48000, amplitude: int = 16000) -> bytes:
+    """Return *ms* milliseconds of a constant-amplitude 16-bit PCM signal."""
+    import math
+
+    n = sample_rate * ms // 1000
+    freq = 440  # Hz
+    samples = [
+        int(amplitude * math.sin(2 * math.pi * freq * i / sample_rate)) for i in range(n)
+    ]
+    return struct.pack(f"<{n}h", *samples)
+
+
+# ── _rms helper ───────────────────────────────────────────────────────────────
+
+
+class TestRmsHelper:
+    """Tests for the internal _rms() energy function."""
+
+    def test_silence_is_zero(self):
+        from integrations.mumble.bridge import _rms
+
+        assert _rms(_pcm_silence()) == 0.0
+
+    def test_empty_bytes_is_zero(self):
+        from integrations.mumble.bridge import _rms
+
+        assert _rms(b"") == 0.0
+
+    def test_tone_has_positive_rms(self):
+        from integrations.mumble.bridge import _rms
+
+        rms = _rms(_pcm_tone(amplitude=16000))
+        assert 0.0 < rms <= 1.0
+
+    def test_louder_tone_has_higher_rms(self):
+        from integrations.mumble.bridge import _rms
+
+        quiet = _rms(_pcm_tone(amplitude=1000))
+        loud = _rms(_pcm_tone(amplitude=20000))
+        assert loud > quiet
+
+    def test_max_amplitude_rms_near_one(self):
+        from integrations.mumble.bridge import _rms
+
+        # All samples at max positive value
+        n = 480
+        pcm = struct.pack(f"<{n}h", *([32767] * n))
+        rms = _rms(pcm)
+        assert rms > 0.99
+
+
+# ── MumbleBridge unit tests ───────────────────────────────────────────────────
+
+
+class TestMumbleBridgeProperties:
+    def test_initial_state(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        assert not bridge.connected
+        assert not bridge.running
+
+    def test_singleton_exists(self):
+        from integrations.mumble.bridge import MumbleBridge, mumble_bridge
+
+        assert isinstance(mumble_bridge, MumbleBridge)
+
+
+class TestMumbleBridgeStart:
+    def test_start_disabled_returns_false(self):
+        """start() returns False when MUMBLE_ENABLED=false."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        mock_settings = MagicMock()
+        mock_settings.mumble_enabled = False
+        with patch("config.settings", mock_settings):
+            result = bridge.start()
+        assert result is False
+        assert not bridge.connected
+
+    def test_start_missing_pymumble_returns_false(self):
+        """start() returns False gracefully when pymumble_py3 is absent."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        mock_settings = MagicMock()
+        mock_settings.mumble_enabled = True
+        with (
+            patch("config.settings", mock_settings),
+            patch.dict(sys.modules, {"pymumble_py3": None}),
+        ):
+            result = bridge.start()
+        assert result is False
+        assert not bridge.connected
+
+    def test_start_already_connected_returns_true(self):
+        """start() short-circuits when already connected."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._connected = True
+        mock_settings = MagicMock()
+        mock_settings.mumble_enabled = True
+        with patch("config.settings", mock_settings):
+            result = bridge.start()
+        assert result is True
+
+    def test_start_connection_error_returns_false(self):
+        """start() returns False and stays clean when Mumble raises."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        mock_settings = MagicMock()
+        mock_settings.mumble_enabled = True
+        mock_settings.mumble_host = "127.0.0.1"
+        mock_settings.mumble_port = 64738
+        mock_settings.mumble_user = "Timmy"
+        mock_settings.mumble_password = ""
+
+        mock_mumble_module = MagicMock()
+        mock_mumble_module.Mumble.side_effect = ConnectionRefusedError("refused")
+
+        with (
+            patch("config.settings", mock_settings),
+            patch.dict(sys.modules, {"pymumble_py3": mock_mumble_module}),
+        ):
+            result = bridge.start()
+
+        assert result is False
+        assert not bridge.connected
+        assert bridge._client is None
+
+
+class TestMumbleBridgeStop:
+    def test_stop_when_not_connected_is_noop(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge.stop()  # Must not raise
+        assert not bridge.connected
+        assert not bridge.running
+
+    def test_stop_clears_state(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._connected = True
+        bridge._running = True
+        mock_client = MagicMock()
+        bridge._client = mock_client
+
+        bridge.stop()
+
+        mock_client.stop.assert_called_once()
+        assert not bridge.connected
+        assert not bridge.running
+        assert bridge._client is None
+
+    def test_stop_tolerates_client_error(self):
+        """stop() cleans up state even when client.stop() raises."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._connected = True
+        bridge._running = True
+        mock_client = MagicMock()
+        mock_client.stop.side_effect = RuntimeError("already stopped")
+        bridge._client = mock_client
+
+        bridge.stop()  # Must not propagate
+
+        assert not bridge.connected
+
+
+# ── Audio send ────────────────────────────────────────────────────────────────
+
+
+class TestMumbleBridgeSendAudio:
+    def test_send_audio_when_not_connected_is_noop(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        pcm = _pcm_tone()
+        bridge.send_audio(pcm)  # Must not raise
+
+    def test_send_audio_enqueues_data(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._connected = True
+        bridge._client = MagicMock()
+
+        pcm = _pcm_tone(ms=20)
+        bridge.send_audio(pcm)
+
+        assert len(bridge._audio_queue) == 1
+        assert bridge._audio_queue[0] == pcm
+
+    def test_send_audio_multiple_chunks(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._connected = True
+        bridge._client = MagicMock()
+
+        for _ in range(3):
+            bridge.send_audio(_pcm_tone(ms=10))
+
+        assert len(bridge._audio_queue) == 3
+
+
+# ── Audio callbacks ───────────────────────────────────────────────────────────
+
+
+class TestMumbleBridgeAudioCallbacks:
+    def test_add_and_trigger_callback(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        received: list[tuple[str, bytes]] = []
+
+        def cb(username: str, pcm: bytes):
+            received.append((username, pcm))
+
+        bridge.add_audio_callback(cb)
+
+        # Simulate sound received
+        fake_user = {"name": "Alexander"}
+        fake_chunk = MagicMock()
+        fake_chunk.pcm = _pcm_tone()
+        bridge._on_sound_received(fake_user, fake_chunk)
+
+        assert len(received) == 1
+        assert received[0][0] == "Alexander"
+
+    def test_remove_callback(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        received: list = []
+
+        def cb(username: str, pcm: bytes):
+            received.append(username)
+
+        bridge.add_audio_callback(cb)
+        bridge.remove_audio_callback(cb)
+
+        fake_user = {"name": "Alexander"}
+        fake_chunk = MagicMock()
+        fake_chunk.pcm = _pcm_tone()
+        bridge._on_sound_received(fake_user, fake_chunk)
+
+        assert received == []
+
+    def test_remove_nonexistent_callback_is_noop(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+
+        def cb(u, p):
+            pass
+
+        bridge.remove_audio_callback(cb)  # Must not raise
+
+    def test_on_sound_received_no_callbacks(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        fake_user = {"name": "Test"}
+        fake_chunk = MagicMock()
+        fake_chunk.pcm = _pcm_tone()
+        bridge._on_sound_received(fake_user, fake_chunk)  # Must not raise
+
+    def test_on_sound_received_missing_user_key(self):
+        """Falls back to 'unknown' when user dict has no 'name' key."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        received_names: list[str] = []
+
+        bridge = MumbleBridge()
+        bridge.add_audio_callback(lambda u, p: received_names.append(u))
+
+        fake_chunk = MagicMock()
+        fake_chunk.pcm = _pcm_tone()
+        bridge._on_sound_received({}, fake_chunk)
+
+        assert received_names == ["unknown"]
+
+    def test_callback_exception_does_not_propagate(self):
+        """A crashing callback must not bubble up to the Mumble thread."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+
+        def bad_cb(u, p):
+            raise RuntimeError("oops")
+
+        bridge.add_audio_callback(bad_cb)
+
+        fake_chunk = MagicMock()
+        fake_chunk.pcm = _pcm_tone()
+        bridge._on_sound_received({"name": "X"}, fake_chunk)  # Must not raise
+
+
+# ── Push-to-talk ──────────────────────────────────────────────────────────────
+
+
+class TestPushToTalk:
+    def test_ptt_context_sets_and_clears_flag(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        assert not bridge._ptt_active
+
+        with bridge.push_to_talk():
+            assert bridge._ptt_active
+
+        assert not bridge._ptt_active
+
+    def test_ptt_clears_on_exception(self):
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        try:
+            with bridge.push_to_talk():
+                raise ValueError("test")
+        except ValueError:
+            pass
+
+        assert not bridge._ptt_active
+
+
+# ── VAD send_pcm_buffer ───────────────────────────────────────────────────────
+
+
+class TestSendPcmBuffer:
+    def test_vad_suppresses_silence(self):
+        """VAD mode must not call sound_output.add_sound for silent PCM."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        mock_client = MagicMock()
+        bridge._client = mock_client
+
+        mock_settings = MagicMock()
+        mock_settings.mumble_audio_mode = "vad"
+        mock_settings.mumble_vad_threshold = 0.02
+
+        with patch("config.settings", mock_settings):
+            bridge._send_pcm_buffer(_pcm_silence(ms=50))
+
+        mock_client.sound_output.add_sound.assert_not_called()
+
+    def test_vad_transmits_tone(self):
+        """VAD mode must send audible PCM frames."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        mock_client = MagicMock()
+        bridge._client = mock_client
+
+        mock_settings = MagicMock()
+        mock_settings.mumble_audio_mode = "vad"
+        mock_settings.mumble_vad_threshold = 0.01
+
+        with patch("config.settings", mock_settings):
+            bridge._send_pcm_buffer(_pcm_tone(ms=50, amplitude=16000))
+
+        assert mock_client.sound_output.add_sound.call_count > 0
+
+    def test_ptt_suppresses_when_inactive(self):
+        """PTT mode must not send when _ptt_active is False."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        mock_client = MagicMock()
+        bridge._client = mock_client
+        bridge._ptt_active = False
+
+        mock_settings = MagicMock()
+        mock_settings.mumble_audio_mode = "ptt"
+        mock_settings.mumble_vad_threshold = 0.02
+
+        with patch("config.settings", mock_settings):
+            bridge._send_pcm_buffer(_pcm_tone(ms=50, amplitude=16000))
+
+        mock_client.sound_output.add_sound.assert_not_called()
+
+    def test_ptt_sends_when_active(self):
+        """PTT mode must send when _ptt_active is True."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        mock_client = MagicMock()
+        bridge._client = mock_client
+        bridge._ptt_active = True
+
+        mock_settings = MagicMock()
+        mock_settings.mumble_audio_mode = "ptt"
+        mock_settings.mumble_vad_threshold = 0.02
+
+        with patch("config.settings", mock_settings):
+            bridge._send_pcm_buffer(_pcm_tone(ms=50, amplitude=16000))
+
+        assert mock_client.sound_output.add_sound.call_count > 0
+
+    def test_no_client_is_noop(self):
+        """_send_pcm_buffer is a no-op when client is None."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._client = None
+        bridge._send_pcm_buffer(_pcm_tone(ms=20))  # Must not raise
+
+
+# ── TTS pipeline ──────────────────────────────────────────────────────────────
+
+
+class TestTtsToPcm:
+    def test_no_tts_engines_returns_none(self):
+        """_tts_to_pcm returns None gracefully when no engine is available."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        with (
+            patch.dict(sys.modules, {"piper": None, "piper.voice": None, "pyttsx3": None}),
+        ):
+            result = bridge._tts_to_pcm("Hello world")
+        assert result is None
+
+    def test_speak_when_not_connected_is_noop(self):
+        """speak() must be a safe no-op when bridge is not connected."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._connected = False
+        bridge.speak("Hello")  # Must not raise
+
+    def test_speak_calls_send_audio_when_tts_succeeds(self):
+        """speak() calls send_audio when _tts_to_pcm returns bytes."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._connected = True
+        bridge._client = MagicMock()
+
+        fake_pcm = _pcm_tone(ms=200)
+        with patch.object(bridge, "_tts_to_pcm", return_value=fake_pcm):
+            with patch.object(bridge, "send_audio") as mock_send:
+                bridge.speak("Hello Timmy")
+
+        mock_send.assert_called_once_with(fake_pcm)
+
+    def test_speak_does_not_call_send_when_tts_fails(self):
+        """speak() does not call send_audio when TTS returns None."""
+        from integrations.mumble.bridge import MumbleBridge
+
+        bridge = MumbleBridge()
+        bridge._connected = True
+        bridge._client = MagicMock()
+
+        with patch.object(bridge, "_tts_to_pcm", return_value=None):
+            with patch.object(bridge, "send_audio") as mock_send:
+                bridge.speak("Hello")
+
+        mock_send.assert_not_called()
+
+
+# ── Config settings integration ───────────────────────────────────────────────
+
+
+class TestMumbleSettings:
+    def test_settings_have_mumble_fields(self):
+        """Settings object exposes all required Mumble configuration fields."""
+        from config import settings
+
+        assert hasattr(settings, "mumble_enabled")
+        assert hasattr(settings, "mumble_host")
+        assert hasattr(settings, "mumble_port")
+        assert hasattr(settings, "mumble_user")
+        assert hasattr(settings, "mumble_password")
+        assert hasattr(settings, "mumble_channel")
+        assert hasattr(settings, "mumble_audio_mode")
+        assert hasattr(settings, "mumble_vad_threshold")
+        assert hasattr(settings, "mumble_silence_ms")
+
+    def test_default_mumble_disabled(self):
+        """Mumble is disabled by default (opt-in only)."""
+        from config import settings
+
+        assert settings.mumble_enabled is False
+
+    def test_default_mumble_port(self):
+        from config import settings
+
+        assert settings.mumble_port == 64738
+
+    def test_default_audio_mode(self):
+        from config import settings
+
+        assert settings.mumble_audio_mode == "vad"
+
+    def test_default_vad_threshold(self):
+        from config import settings
+
+        assert 0.0 < settings.mumble_vad_threshold < 1.0