diff --git a/tests/tools/test_voice_mode.py b/tests/tools/test_voice_mode.py index 4f0b31d9905..fdab4e8d8eb 100644 --- a/tests/tools/test_voice_mode.py +++ b/tests/tools/test_voice_mode.py @@ -265,6 +265,46 @@ class TestDetectAudioEnvironment: assert result["warnings"] == [] assert any("container" in n.lower() for n in result.get("notices", [])) + def test_docker_with_pipewire_remote_and_no_devices_allows_voice(self, monkeypatch): + """PIPEWIRE_REMOTE should bypass empty PortAudio device lists in Docker.""" + monkeypatch.delenv("SSH_CLIENT", raising=False) + monkeypatch.delenv("SSH_TTY", raising=False) + monkeypatch.delenv("SSH_CONNECTION", raising=False) + monkeypatch.delenv("PULSE_SERVER", raising=False) + monkeypatch.setenv("PIPEWIRE_REMOTE", "/run/user/1000/pipewire-0") + monkeypatch.setattr("hermes_constants.is_container", lambda: True) + + sd = MagicMock() + sd.query_devices.return_value = [] + monkeypatch.setattr("tools.voice_mode._import_audio", lambda: (sd, MagicMock())) + + from tools.voice_mode import detect_audio_environment + result = detect_audio_environment() + + assert result["available"] is True + assert result["warnings"] == [] + assert any("host audio forwarding" in n.lower() for n in result.get("notices", [])) + + def test_docker_with_pipewire_remote_and_query_failure_allows_voice(self, monkeypatch): + """PIPEWIRE_REMOTE should bypass PortAudio query failures in Docker.""" + monkeypatch.delenv("SSH_CLIENT", raising=False) + monkeypatch.delenv("SSH_TTY", raising=False) + monkeypatch.delenv("SSH_CONNECTION", raising=False) + monkeypatch.delenv("PULSE_SERVER", raising=False) + monkeypatch.setenv("PIPEWIRE_REMOTE", "/run/user/1000/pipewire-0") + monkeypatch.setattr("hermes_constants.is_container", lambda: True) + + sd = MagicMock() + sd.query_devices.side_effect = RuntimeError("boom") + monkeypatch.setattr("tools.voice_mode._import_audio", lambda: (sd, MagicMock())) + + from tools.voice_mode import detect_audio_environment + result = detect_audio_environment() + + assert result["available"] is True + assert result["warnings"] == [] + assert any("host audio forwarding" in n.lower() for n in result.get("notices", [])) + def test_docker_without_audio_forwarding_blocks_voice(self, monkeypatch): """Docker without PULSE_SERVER/PIPEWIRE_REMOTE keeps blocking voice mode.""" monkeypatch.delenv("SSH_CLIENT", raising=False) diff --git a/tools/voice_mode.py b/tools/voice_mode.py index 0ba449d87ae..e98fcef8857 100644 --- a/tools/voice_mode.py +++ b/tools/voice_mode.py @@ -97,6 +97,9 @@ def detect_audio_environment() -> dict: termux_mic_cmd = _termux_microphone_command() termux_app_installed = _termux_api_app_installed() termux_capture = bool(termux_mic_cmd and termux_app_installed) + has_forwarded_audio = bool( + os.environ.get('PULSE_SERVER') or os.environ.get('PIPEWIRE_REMOTE') + ) # SSH detection if any(os.environ.get(v) for v in ('SSH_CLIENT', 'SSH_TTY', 'SSH_CONNECTION')): @@ -108,7 +111,7 @@ def detect_audio_environment() -> dict: # (issue #21203). Only block when no forwarding is configured. from hermes_constants import is_container if is_container(): - if os.environ.get('PULSE_SERVER') or os.environ.get('PIPEWIRE_REMOTE'): + if has_forwarded_audio: notices.append("Running inside container (Docker/Podman/LXC) with host audio forwarding") else: warnings.append( @@ -143,17 +146,22 @@ def detect_audio_environment() -> dict: try: devices = sd.query_devices() if not devices: - if os.environ.get('PULSE_SERVER'): - notices.append("No PortAudio devices detected but PULSE_SERVER is set -- continuing") + if has_forwarded_audio: + notices.append( + "No PortAudio devices detected but host audio forwarding is configured -- continuing" + ) elif termux_capture: notices.append("No PortAudio devices detected, but Termux:API microphone capture is available") else: warnings.append("No audio input/output devices detected") except Exception: # In WSL with PulseAudio, device queries can fail even though - # recording/playback works fine. Don't block if PULSE_SERVER is set. - if os.environ.get('PULSE_SERVER'): - notices.append("Audio device query failed but PULSE_SERVER is set -- continuing") + # recording/playback works fine. Don't block if host audio + # forwarding is configured. + if has_forwarded_audio: + notices.append( + "Audio device query failed but host audio forwarding is configured -- continuing" + ) elif termux_capture: notices.append("PortAudio device query failed, but Termux:API microphone capture is available") else: