test: adapt salvaged voice tests to current main + lint fix

- _FakeProc gains returncode (main's player loop checks proc.returncode)
- WSL gate tests clear SSH_* env vars (main hard-warns over SSH without
  forwarded audio) and accept the merged #37346 forwarded-sound-server
  notice wording
- test_tts_macos_output stubs resolve_streaming_provider so the
  OutputStream setup path actually runs on main's chunked-streamer code
- voice CLI integration tests unwrap the _VoiceInputMessage sentinel
- _is_wsl: explicit encoding + drop unreachable return (ruff PLW1514)
This commit is contained in:
Teknium 2026-07-28 09:32:03 -07:00
parent fe3dc29009
commit e6b0344052
4 changed files with 42 additions and 5 deletions

View file

@ -12,6 +12,16 @@ import threading
import pytest
class _FakeStreamer:
"""Minimal chunked-streamer stand-in so the OutputStream setup path runs."""
sample_rate = 24000
channels = 1
def stream(self, text):
return iter([])
def _run_stream(monkeypatch, system_name):
"""Drive stream_tts_to_speaker once with a mock client on *system_name*.
@ -33,6 +43,10 @@ def _run_stream(monkeypatch, system_name):
return iter([]) # no audio chunks needed for setup assertion
monkeypatch.setattr("tools.tts_tool._import_elevenlabs", lambda: _FakeTTS)
monkeypatch.setattr(
"tools.tts_streaming.resolve_streaming_provider",
lambda cfg, preferred=None: _FakeStreamer(),
)
sd_called = {"hit": False}
@ -81,6 +95,10 @@ def _run_stream_offmac(monkeypatch):
return iter([])
monkeypatch.setattr("tools.tts_tool._import_elevenlabs", lambda: _FakeTTS)
monkeypatch.setattr(
"tools.tts_streaming.resolve_streaming_provider",
lambda cfg, preferred=None: _FakeStreamer(),
)
sd_called = {"hit": False}

View file

@ -1208,7 +1208,12 @@ class TestVoiceStopAndTranscribeReal:
recorder.stop.return_value = "/tmp/test.wav"
cli = _make_voice_cli(_voice_recording=True, _voice_recorder=recorder)
cli._voice_stop_and_transcribe()
assert cli._pending_input.get_nowait() == "hello world"
queued = cli._pending_input.get_nowait()
# Voice transcripts are wrapped in the _VoiceInputMessage sentinel so
# only genuine STT output gets the voice prefix (#65827).
from cli import _VoiceInputMessage
assert isinstance(queued, _VoiceInputMessage)
assert str(queued) == "hello world"
@patch("cli._cprint")
@patch("cli.os.unlink")
@ -1424,7 +1429,10 @@ class TestVoiceBargeCaptureSubmit:
cli._voice_submit_barge_utterance(str(wav))
assert cli._pending_input.get_nowait() == "stop, do it differently"
queued = cli._pending_input.get_nowait()
from cli import _VoiceInputMessage
assert isinstance(queued, _VoiceInputMessage)
assert str(queued) == "stop, do it differently"
assert not cli._voice_barge_capture.is_set()
assert not wav.exists()

View file

@ -1243,6 +1243,8 @@ class TestMacOSAudioOutputPolicy:
popen_cmds = []
class _FakeProc:
returncode = 0
def wait(self, timeout=None):
return 0
@ -2437,6 +2439,8 @@ class TestWSLAudioEnvironmentGate:
monkeypatch.delenv("PULSE_SERVER", raising=False)
monkeypatch.delenv("PIPEWIRE_REMOTE", raising=False)
for _ssh_var in ("SSH_CLIENT", "SSH_TTY", "SSH_CONNECTION"):
monkeypatch.delenv(_ssh_var, raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
@ -2462,6 +2466,8 @@ class TestWSLAudioEnvironmentGate:
monkeypatch.delenv("PULSE_SERVER", raising=False)
monkeypatch.delenv("PIPEWIRE_REMOTE", raising=False)
for _ssh_var in ("SSH_CLIENT", "SSH_TTY", "SSH_CONNECTION"):
monkeypatch.delenv(_ssh_var, raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
@ -2481,6 +2487,8 @@ class TestWSLAudioEnvironmentGate:
from tools import voice_mode as vm
monkeypatch.setenv("PULSE_SERVER", "unix:/mnt/wslg/PulseServer")
for _ssh_var in ("SSH_CLIENT", "SSH_TTY", "SSH_CONNECTION"):
monkeypatch.delenv(_ssh_var, raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio",
lambda: (MagicMock(), MagicMock()))
with patch("builtins.open", side_effect=self._fake_open_wsl), \
@ -2488,4 +2496,8 @@ class TestWSLAudioEnvironmentGate:
result = vm.detect_audio_environment()
assert result["available"] is True
assert any("PulseAudio bridge" in n for n in result["notices"])
# Merged with #37346: any forwarded sound server (PULSE_SERVER or
# PIPEWIRE_REMOTE) yields the shared reachable-sound-server notice.
assert any(
"PulseAudio" in n and "WSL" in n for n in result["notices"]
)

View file

@ -1341,11 +1341,10 @@ def stop_playback() -> None:
def _is_wsl() -> bool:
"""True when running inside Windows Subsystem for Linux."""
try:
with open("/proc/version", "r") as f:
with open("/proc/version", "r", encoding="utf-8", errors="replace") as f:
return "microsoft" in f.read().lower()
except Exception:
return False
return False
def _is_wsl2_env() -> bool: