From c5fe636cd08011993281cd49c175693efcd3e7eb Mon Sep 17 00:00:00 2001 From: StuBehan Date: Wed, 29 Apr 2026 13:22:38 +0200 Subject: [PATCH] test: bring coverage from 50% to 86% MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds tests for the engine, CLI subcommand handlers, and the non-protocol parts of the daemon — all mocked at the right boundary so no real model load or audio device is required. - tests/test_engine.py (new): Stackvox.__init__/synthesize/speak/voices/ speak_sequence, the module-level speak/synthesize singletons, the thread-safety contract on _get_default, the urlretrieve reporthook progress output, and _ensure_models's skip-existing/download-missing logic. Mocks Kokoro and sounddevice so nothing runtime-heavy runs. - tests/test_cli.py (extended): every _cmd_* handler — speak (text/--out/ blank), say (daemon-up/down/fallback-say with-and-without `say` binary), serve (normal + RuntimeError), stop (running/stopped), status, voices, welcome, completion (defensive unsupported-shell branch). - tests/test_daemon.py (new): _pid_alive against this process and an unused PID, is_running across missing/garbage/alive/dead pid files, the send/say/stop/ping client wrappers (payload shape + timeout), _refresh_audio_devices happy-path and exception swallowing, and the _start_device_watcher non-darwin no-op. Coverage by file: cli.py 54% → 99% engine.py 26% → 93% daemon.py 55% → 73% (remainder is the macOS CoreAudio ctypes watcher, hard to test portably) overall 50% → 86% 69 tests, all green. ruff + mypy clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_cli.py | 162 +++++++++++++++++++++++++++++ tests/test_daemon.py | 138 +++++++++++++++++++++++++ tests/test_engine.py | 241 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 541 insertions(+) create mode 100644 tests/test_daemon.py create mode 100644 tests/test_engine.py diff --git a/tests/test_cli.py b/tests/test_cli.py index d4a9577..6d4d041 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -89,3 +89,165 @@ def test_completion_bash_emits_complete_script(mocker, capsys): captured = capsys.readouterr() assert "_stackvox_completion()" in captured.out assert "complete -F _stackvox_completion stackvox" in captured.out + + +# --- Subcommand handlers ---------------------------------------------------- +# +# Each subcommand handler is exercised directly with a constructed Namespace +# so we don't have to reason about argparse behavior for every test. The +# Stackvox class and the daemon module's surface are mocked out. + + +@pytest.fixture +def fake_stackvox(mocker): + return mocker.patch.object(cli, "Stackvox") + + +def _ns(**kwargs): + """Build a minimal argparse.Namespace for handler tests.""" + import argparse + + return argparse.Namespace(**kwargs) + + +class TestCmdSpeak: + def test_with_text_calls_engine_speak(self, fake_stackvox): + rc = cli._cmd_speak(_ns(voice="af_sarah", speed=1.0, lang="en-us", text="hello", out=None)) + assert rc == 0 + fake_stackvox.return_value.speak.assert_called_once_with("hello") + + def test_with_out_writes_wav_instead_of_playing(self, fake_stackvox, mocker, tmp_path): + import numpy as np + + fake_stackvox.return_value.synthesize.return_value = ( + np.zeros(10, dtype=np.float32), + 24000, + ) + sf_write = mocker.patch.object(cli.sf, "write") + out = tmp_path / "out.wav" + + rc = cli._cmd_speak(_ns(voice="af_sarah", speed=1.0, lang="en-us", text="hi", out=out)) + + assert rc == 0 + sf_write.assert_called_once() + fake_stackvox.return_value.speak.assert_not_called() + + def test_blank_input_returns_error(self, fake_stackvox, capsys): + rc = cli._cmd_speak(_ns(voice="af_sarah", speed=1.0, lang="en-us", text=" ", out=None)) + assert rc == 1 + assert "provide text" in capsys.readouterr().err + fake_stackvox.return_value.speak.assert_not_called() + + +class TestCmdSay: + def test_returns_zero_when_daemon_accepts(self, mocker): + mocker.patch.object(cli.daemon, "say", return_value=(True, "ok")) + rc = cli._cmd_say(_ns(voice="af_sarah", speed=1.0, lang="en-us", text="hi", fallback_say=False)) + assert rc == 0 + + def test_returns_two_when_daemon_unreachable_and_no_fallback(self, mocker, capsys): + mocker.patch.object(cli.daemon, "say", return_value=(False, "daemon not running")) + rc = cli._cmd_say(_ns(voice="af_sarah", speed=1.0, lang="en-us", text="hi", fallback_say=False)) + assert rc == 2 + assert "daemon not running" in capsys.readouterr().err + + def test_fallback_say_shells_out_on_macos(self, mocker): + mocker.patch.object(cli.daemon, "say", return_value=(False, "daemon not running")) + # Must use which because cli imports inside the function body. + mocker.patch("shutil.which", return_value="/usr/bin/say") + run = mocker.patch("subprocess.run") + rc = cli._cmd_say(_ns(voice="af_sarah", speed=1.0, lang="en-us", text="hi", fallback_say=True)) + assert rc == 0 + run.assert_called_once() + assert run.call_args.args[0][0] == "say" + + def test_fallback_say_without_say_binary_returns_two(self, mocker, capsys): + mocker.patch.object(cli.daemon, "say", return_value=(False, "daemon not running")) + mocker.patch("shutil.which", return_value=None) + rc = cli._cmd_say(_ns(voice="af_sarah", speed=1.0, lang="en-us", text="hi", fallback_say=True)) + assert rc == 2 + + def test_blank_input_returns_one(self, capsys): + rc = cli._cmd_say(_ns(voice="af_sarah", speed=1.0, lang="en-us", text="", fallback_say=False)) + assert rc == 1 + + +class TestCmdServe: + def test_propagates_serve_args(self, mocker): + serve = mocker.patch.object(cli.daemon, "serve") + rc = cli._cmd_serve(_ns(voice="bf_emma", speed=1.1, lang="en-gb")) + assert rc == 0 + serve.assert_called_once_with(voice="bf_emma", speed=1.1, lang="en-gb") + + def test_returns_one_when_daemon_already_running(self, mocker, capsys): + mocker.patch.object(cli.daemon, "serve", side_effect=RuntimeError("daemon already running")) + rc = cli._cmd_serve(_ns(voice="af_sarah", speed=1.0, lang="en-us")) + assert rc == 1 + assert "already running" in capsys.readouterr().err + + +class TestCmdStop: + def test_returns_zero_when_daemon_already_stopped(self, mocker, capsys): + mocker.patch.object(cli.daemon, "is_running", return_value=False) + stop = mocker.patch.object(cli.daemon, "stop") + rc = cli._cmd_stop(_ns()) + assert rc == 0 + stop.assert_not_called() + assert "not running" in capsys.readouterr().err + + def test_calls_daemon_stop_when_running(self, mocker): + mocker.patch.object(cli.daemon, "is_running", return_value=True) + mocker.patch.object(cli.daemon, "stop", return_value=(True, "ok")) + assert cli._cmd_stop(_ns()) == 0 + + +class TestCmdStatus: + def test_running_prints_pid_and_returns_zero(self, mocker, capsys): + mocker.patch.object(cli.daemon, "is_running", return_value=True) + # PID_PATH and SOCKET_PATH are read at module level — mock as needed. + pid_path = mocker.MagicMock() + pid_path.read_text.return_value = "12345\n" + mocker.patch.object(cli.daemon, "PID_PATH", pid_path) + mocker.patch.object(cli.daemon, "SOCKET_PATH", "/tmp/x.sock") + + rc = cli._cmd_status(_ns()) + + assert rc == 0 + out = capsys.readouterr().out + assert "running" in out + assert "12345" in out + + def test_stopped_returns_one(self, mocker, capsys): + mocker.patch.object(cli.daemon, "is_running", return_value=False) + rc = cli._cmd_status(_ns()) + assert rc == 1 + assert "stopped" in capsys.readouterr().out + + +class TestCmdVoices: + def test_prints_one_voice_per_line(self, fake_stackvox, capsys): + fake_stackvox.return_value.voices.return_value = ["af_sarah", "bf_emma"] + rc = cli._cmd_voices(_ns()) + assert rc == 0 + assert capsys.readouterr().out.split() == ["af_sarah", "bf_emma"] + + +class TestCmdWelcome: + def test_calls_speak_sequence_with_welcome_lines(self, fake_stackvox): + cli._cmd_welcome(_ns()) + fake_stackvox.return_value.speak_sequence.assert_called_once() + lines = fake_stackvox.return_value.speak_sequence.call_args.args[0] + # All WELCOME_LINES rows present. + assert len(lines) == len(cli.WELCOME_LINES) + # Every entry has text/voice/lang. + for entry in lines: + assert {"text", "voice", "lang"} <= entry.keys() + + +class TestCmdCompletion: + def test_unsupported_shell_returns_one(self, capsys): + # argparse choices=["bash"] means we'd never reach here normally; + # exercise the defensive branch directly. + rc = cli._cmd_completion(_ns(shell="fish")) + assert rc == 1 + assert "unsupported shell" in capsys.readouterr().err diff --git a/tests/test_daemon.py b/tests/test_daemon.py new file mode 100644 index 0000000..d7c5eb9 --- /dev/null +++ b/tests/test_daemon.py @@ -0,0 +1,138 @@ +"""Daemon tests for non-protocol surface — pid/socket helpers, send/say/stop/ping +client, _refresh_audio_devices, and the cross-platform watcher entry point. +The protocol/handler tests live in test_daemon_protocol.py. +""" + +from __future__ import annotations + +import os + +import pytest + +from stackvox import daemon + + +class TestPidAlive: + def test_self_pid_is_alive(self): + assert daemon._pid_alive(os.getpid()) is True + + def test_unused_pid_is_not_alive(self): + # PID 0 is reserved on macOS/Linux and never owned by a real process. + # `os.kill(0, 0)` is special-cased (signals the whole process group), + # so use a high PID we can be confident is unused. + assert daemon._pid_alive(2**31 - 1) is False + + +class TestIsRunning: + def test_returns_false_when_pid_file_missing(self, mocker, tmp_path): + mocker.patch.object(daemon, "PID_PATH", tmp_path / "missing.pid") + assert daemon.is_running() is False + + def test_returns_false_when_pid_file_unreadable(self, mocker, tmp_path): + pid = tmp_path / "garbage.pid" + pid.write_text("not-a-number") + mocker.patch.object(daemon, "PID_PATH", pid) + assert daemon.is_running() is False + + def test_returns_true_when_pid_alive(self, mocker, tmp_path): + pid = tmp_path / "live.pid" + pid.write_text(str(os.getpid())) + mocker.patch.object(daemon, "PID_PATH", pid) + assert daemon.is_running() is True + + def test_returns_false_when_pid_dead(self, mocker, tmp_path): + pid = tmp_path / "dead.pid" + pid.write_text(str(2**31 - 1)) + mocker.patch.object(daemon, "PID_PATH", pid) + assert daemon.is_running() is False + + +class TestSendHelpers: + """The thin send/say/stop/ping wrappers around the unix-socket protocol.""" + + def test_send_returns_failure_when_socket_missing(self, mocker, tmp_path): + mocker.patch.object(daemon, "SOCKET_PATH", tmp_path / "nope.sock") + ok, resp = daemon.send({"command": "ping"}) + assert ok is False + assert resp == "daemon not running" + + def test_say_includes_only_supplied_overrides(self, mocker): + send = mocker.patch.object(daemon, "send", return_value=(True, "ok")) + daemon.say("hello", voice="bf_emma") + payload = send.call_args.args[0] + assert payload == {"text": "hello", "voice": "bf_emma"} + + def test_say_with_all_overrides(self, mocker): + send = mocker.patch.object(daemon, "send", return_value=(True, "ok")) + daemon.say("hi", voice="af_sarah", speed=1.5, lang="en-us") + assert send.call_args.args[0] == { + "text": "hi", + "voice": "af_sarah", + "speed": 1.5, + "lang": "en-us", + } + + def test_say_without_overrides_passes_only_text(self, mocker): + send = mocker.patch.object(daemon, "send", return_value=(True, "ok")) + daemon.say("hi") + assert send.call_args.args[0] == {"text": "hi"} + + def test_stop_sends_command_stop(self, mocker): + send = mocker.patch.object(daemon, "send", return_value=(True, "ok")) + daemon.stop() + assert send.call_args.args[0] == {"command": "stop"} + + def test_ping_uses_short_timeout(self, mocker): + send = mocker.patch.object(daemon, "send", return_value=(True, "ok")) + daemon.ping() + assert send.call_args.kwargs["timeout"] == daemon.PING_TIMEOUT_SECONDS + + +class TestRefreshAudioDevices: + def test_calls_terminate_then_initialize(self, mocker): + terminate = mocker.patch.object(daemon.sd, "_terminate") + initialize = mocker.patch.object(daemon.sd, "_initialize") + + daemon._refresh_audio_devices() + + terminate.assert_called_once() + initialize.assert_called_once() + + def test_swallows_and_logs_errors(self, mocker, caplog): + """A failure in the audio reset must not propagate out — the worker + retries on the next request.""" + import logging + + mocker.patch.object(daemon.sd, "_terminate", side_effect=RuntimeError("portaudio gone")) + mocker.patch.object(daemon.sd, "_initialize") + + with caplog.at_level(logging.ERROR, logger="stackvox.daemon"): + daemon._refresh_audio_devices() # must not raise + + assert any("failed to refresh audio devices" in r.message for r in caplog.records) + + +class TestStartDeviceWatcher: + def test_no_op_on_non_darwin(self, mocker): + """On Linux/Windows the function returns immediately; nothing should + be loaded from CoreAudio and no thread should be started.""" + mocker.patch.object(daemon.sys, "platform", "linux") + cdll = mocker.patch.object(daemon.ctypes, "CDLL") + + daemon._start_device_watcher() + + cdll.assert_not_called() + + @pytest.mark.skipif( + not hasattr(daemon.sys, "platform") or daemon.sys.platform != "darwin", + reason="macOS-only behaviour", + ) + def test_disabled_when_coreaudio_unavailable(self, mocker, caplog): + """If CDLL fails to load CoreAudio (unusual on real macOS but possible + in stripped environments), the watcher should bail without raising.""" + import logging + + mocker.patch.object(daemon.ctypes, "CDLL", side_effect=OSError("no coreaudio")) + with caplog.at_level(logging.DEBUG, logger="stackvox.daemon"): + daemon._start_device_watcher() # must not raise + assert any("CoreAudio unavailable" in r.message for r in caplog.records) diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..e981717 --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,241 @@ +"""Engine tests — no real model load, no audio device. + +We mock at two boundaries: `Kokoro` (so `Stackvox.__init__` doesn't pull the +model) and `sounddevice` (so `speak()` doesn't try to talk to PortAudio). +This lets us verify call-shape and side effects without any of the heavy +runtime dependencies. +""" + +from __future__ import annotations + +import threading + +import numpy as np +import pytest + +from stackvox import engine + + +@pytest.fixture +def fake_ensure_models(mocker, tmp_path): + """Short-circuit _ensure_models so Stackvox.__init__ skips the download.""" + model = tmp_path / "kokoro.onnx" + voices = tmp_path / "voices.bin" + model.touch() + voices.touch() + return mocker.patch.object(engine, "_ensure_models", return_value=(model, voices)) + + +@pytest.fixture +def fake_kokoro(mocker, fake_ensure_models): + """Mock Kokoro on top of fake_ensure_models — most tests want both.""" + return mocker.patch.object(engine, "Kokoro") + + +@pytest.fixture +def fake_audio(mocker): + """Mock the sounddevice surface used by Stackvox.""" + mocker.patch.object(engine.sd, "play") + mocker.patch.object(engine.sd, "wait") + mocker.patch.object(engine.sd, "stop") + + +@pytest.fixture(autouse=True) +def reset_default(): + """Each test starts with no module-level singleton.""" + engine._default = None + yield + engine._default = None + + +class TestStackvoxInit: + def test_stores_voice_speed_lang(self, fake_kokoro): + tts = engine.Stackvox(voice="bf_emma", speed=1.2, lang="en-gb") + assert tts.voice == "bf_emma" + assert tts.speed == 1.2 + assert tts.lang == "en-gb" + + def test_uses_custom_cache_dir(self, fake_kokoro, fake_ensure_models, tmp_path): + custom = tmp_path / "custom-cache" + engine.Stackvox(cache_dir=custom) + fake_ensure_models.assert_called_once_with(custom) + + def test_falls_back_to_default_cache_dir(self, fake_kokoro, fake_ensure_models, mocker, tmp_path): + default = tmp_path / "default" + mocker.patch.object(engine, "_default_cache_dir", return_value=default) + engine.Stackvox() + fake_ensure_models.assert_called_once_with(default) + + +class TestSynthesize: + def test_passes_engine_defaults_to_kokoro(self, fake_kokoro): + samples = np.zeros(10, dtype=np.float32) + fake_kokoro.return_value.create.return_value = (samples, 24000) + + tts = engine.Stackvox(voice="af_sarah", speed=1.0, lang="en-us") + result_samples, result_sr = tts.synthesize("hi") + + fake_kokoro.return_value.create.assert_called_once_with( + "hi", voice="af_sarah", speed=1.0, lang="en-us" + ) + assert result_sr == 24000 + assert np.array_equal(result_samples, samples) + + def test_per_call_overrides_take_priority(self, fake_kokoro): + fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) + tts = engine.Stackvox(voice="af_sarah", speed=1.0, lang="en-us") + tts.synthesize("hi", voice="bf_emma", speed=1.5, lang="en-gb") + fake_kokoro.return_value.create.assert_called_once_with( + "hi", voice="bf_emma", speed=1.5, lang="en-gb" + ) + + def test_speed_zero_override_is_respected(self, fake_kokoro): + """speed=0 is falsy but a valid override; ensure it's not silently dropped.""" + fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) + tts = engine.Stackvox(speed=1.0) + tts.synthesize("hi", speed=0.5) + kwargs = fake_kokoro.return_value.create.call_args.kwargs + assert kwargs["speed"] == 0.5 + + +class TestSpeak: + def test_blocking_calls_play_and_wait(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) + tts = engine.Stackvox() + tts.speak("hi") + engine.sd.play.assert_called_once() + engine.sd.wait.assert_called_once() + + def test_non_blocking_skips_wait(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) + tts = engine.Stackvox() + tts.speak("hi", blocking=False) + engine.sd.play.assert_called_once() + engine.sd.wait.assert_not_called() + + def test_stop_calls_sounddevice_stop(self, fake_kokoro, fake_audio): + engine.Stackvox().stop() + engine.sd.stop.assert_called_once() + + +class TestVoices: + def test_returns_sorted_voice_ids(self, fake_kokoro): + fake_kokoro.return_value.get_voices.return_value = ["bf_emma", "af_sarah", "am_adam"] + tts = engine.Stackvox() + assert tts.voices() == ["af_sarah", "am_adam", "bf_emma"] + + +class TestSpeakSequence: + def test_empty_lines_returns_early(self, fake_kokoro, fake_audio): + engine.Stackvox().speak_sequence([]) + engine.sd.play.assert_not_called() + + def test_concatenates_segments_with_optional_gap(self, fake_kokoro, fake_audio): + a = np.array([1.0, 1.0, 1.0], dtype=np.float32) + b = np.array([2.0, 2.0], dtype=np.float32) + fake_kokoro.return_value.create.side_effect = [(a, 100), (b, 100)] + + tts = engine.Stackvox() + tts.speak_sequence([{"text": "first"}, {"text": "second"}], gap_seconds=0.05, concurrent=False) + + played = engine.sd.play.call_args.args[0] + # Length: 3 + (100 * 0.05) silence + 2 = 10 samples. + assert len(played) == 3 + 5 + 2 + # First 3 samples are `a`, last 2 are `b`, middle 5 are silence. + np.testing.assert_array_equal(played[:3], a) + np.testing.assert_array_equal(played[-2:], b) + np.testing.assert_array_equal(played[3:8], np.zeros(5, dtype=np.float32)) + + +class TestModuleLevelHelpers: + def test_speak_reuses_module_singleton(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) + engine.speak("first") + engine.speak("second") + # Stackvox class only instantiated once → Kokoro only constructed once. + assert fake_kokoro.call_count == 1 + + def test_synthesize_reuses_module_singleton(self, fake_kokoro): + fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) + engine.synthesize("first") + engine.synthesize("second") + assert fake_kokoro.call_count == 1 + + def test_get_default_is_thread_safe(self, fake_kokoro): + """Concurrent first calls must not double-instantiate Stackvox. + + Without the lock around _get_default, multiple threads calling + speak()/synthesize() at process start can both observe `_default is + None` and each construct a Stackvox (each loading the 340 MB model). + """ + fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) + barrier = threading.Barrier(8) + + def race(): + barrier.wait() + engine._get_default() + + threads = [threading.Thread(target=race) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert fake_kokoro.call_count == 1 + + +class TestDownloadProgress: + def test_reporthook_prints_percentage_when_size_known(self, mocker, capsys, tmp_path): + """The hook should print updating progress lines while downloading.""" + captured_hook = {} + + def fake_urlretrieve(url, dest, reporthook=None): + captured_hook["fn"] = reporthook + # Simulate three chunks of a 1000-byte download. + if reporthook: + reporthook(0, 100, 1000) + reporthook(5, 100, 1000) + reporthook(10, 100, 1000) + + mocker.patch.object(engine.urllib.request, "urlretrieve", side_effect=fake_urlretrieve) + engine._download_with_progress("http://example/m.bin", tmp_path / "m.bin") + + err = capsys.readouterr().err + assert "downloading m.bin" in err + assert "100%" in err + + def test_unknown_size_does_not_print(self, mocker, capsys, tmp_path): + """totalsize <= 0 means Content-Length absent; hook should be a no-op.""" + + def fake_urlretrieve(url, dest, reporthook=None): + if reporthook: + reporthook(0, 100, -1) + + mocker.patch.object(engine.urllib.request, "urlretrieve", side_effect=fake_urlretrieve) + engine._download_with_progress("http://example/m.bin", tmp_path / "m.bin") + + # No percentage should have been emitted because total size was unknown. + assert "downloading" not in capsys.readouterr().err + + +class TestEnsureModels: + def test_skips_files_that_already_exist(self, mocker, tmp_path): + download = mocker.patch.object(engine, "_download_with_progress") + (tmp_path / "kokoro-v1.0.onnx").touch() + (tmp_path / "voices-v1.0.bin").touch() + + engine._ensure_models(tmp_path) + + download.assert_not_called() + + def test_downloads_only_missing_files(self, mocker, tmp_path): + download = mocker.patch.object(engine, "_download_with_progress") + # Only the model exists; voices needs downloading. + (tmp_path / "kokoro-v1.0.onnx").touch() + + engine._ensure_models(tmp_path) + + # Exactly one download, for the voices file. + assert download.call_count == 1 + url_arg, dest_arg = download.call_args.args + assert dest_arg.name == "voices-v1.0.bin"