test: bring coverage from 50% to 86%

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

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)