STT-Assistant/tests/test_supervisor.py

"""Tests for the Supervisor class (daemon, assistant, TUI orchestration)."""

import signal
import sys
import types

from stt_daemon.config import Config
from stt_daemon.supervisor import Supervisor


class FakePopen:
    """Fake subprocess.Popen for testing without spawning real processes."""

    _pid_counter = 10000

    def __init__(self, argv, redirect_output=False):
        self.argv = argv
        self.redirect_output = redirect_output
        self.pid = FakePopen._pid_counter
        FakePopen._pid_counter += 1
        self.returncode = None
        self._terminated = False
        self._killed = False

    def poll(self):
        """Return None (running) or a returncode."""
        return self.returncode

    def terminate(self):
        """Simulate termination."""
        self._terminated = True
        if self.returncode is None:
            self.returncode = -15

    def kill(self):
        """Simulate kill."""
        self._killed = True
        if self.returncode is None:
            self.returncode = -9

    def wait(self, timeout=None):
        """Wait for process to exit."""
        if self.returncode is None:
            self.returncode = 0
        return self.returncode


def _cfg():
    """Create a test Config object."""
    return Config()


class FakeTermios:
    """Records tcgetattr/tcsetattr calls instead of touching a real terminal."""

    TCSADRAIN = 1
    ISIG = 1

    def __init__(self, log):
        self._log = log

    def tcgetattr(self, fd):
        self._log.append(("get", fd))
        return [0, 0, 0, 0, 0, 0, []]

    def tcsetattr(self, fd, when, attrs):
        self._log.append(("set", fd, when, attrs))


class FakeStdin:
    """Stand-in for sys.stdin so the helpers never touch pytest's captured stdin."""

    def fileno(self):
        return 7


def test_supervisor_run_spawns_daemon_then_assistant_no_tui():
    """run() spawns daemon then assistant, not TUI when with_tui=False."""
    cfg = _cfg()
    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        spawned.append((argv, redirect_output))
        proc = FakePopen(argv, redirect_output=redirect_output)
        if len(spawned) == 1:
            proc.returncode = 0
        return proc

    supervisor = Supervisor(cfg, with_tui=False, spawn=fake_spawn, poll_s=0.001)
    result = supervisor.run()

    assert len(spawned) == 2
    assert "stt_daemon" in " ".join(spawned[0][0]) and "run" in spawned[0][0]
    assert "stt_daemon.assistant" in " ".join(spawned[1][0])
    assert "stt_daemon.tui" not in " ".join(str(x) for x in spawned)
    assert result == 0


def test_supervisor_run_fails_when_tui_socket_timeout():
    """run() with_tui=True fails and terminates all if control socket timeout."""
    cfg = _cfg()
    cfg.assistant.control_socket = "/nonexistent/socket.sock"

    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        proc = FakePopen(argv, redirect_output=redirect_output)
        spawned.append(proc)
        return proc

    clock_val = [0.0]

    def fake_clock():
        return clock_val[0]

    def fake_sleep(s):
        clock_val[0] += 0.1

    supervisor = Supervisor(
        cfg,
        with_tui=True,
        spawn=fake_spawn,
        sleep=fake_sleep,
        clock=fake_clock,
        poll_s=0.001,
        wait_timeout_s=0.05,
    )
    result = supervisor.run()

    assert result == 1
    assert len(spawned) == 2
    for proc in spawned:
        assert proc._terminated or proc._killed


def test_supervisor_run_returns_child_exit_code():
    """run() returns first child's exit code when it exits."""
    cfg = _cfg()
    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        proc = FakePopen(argv, redirect_output=redirect_output)
        spawned.append(proc)
        if len(spawned) == 1:
            proc.returncode = 42
        return proc

    supervisor = Supervisor(cfg, with_tui=False, spawn=fake_spawn, poll_s=0.001)
    result = supervisor.run()

    assert result == 42
    assert spawned[0].returncode == 42
    assert spawned[1]._terminated or spawned[1]._killed


def test_supervisor_forwards_overrides_to_daemon():
    """Daemon receives device, backend, model, language overrides."""
    cfg = _cfg()
    overrides = {
        "device": "microphone.usb",
        "backend": "sounddevice",
        "model": "base",
        "language": "es",
        "config": "/path/to/config.toml",
    }
    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        spawned.append(argv)
        proc = FakePopen(argv, redirect_output=redirect_output)
        if len(spawned) == 1:
            proc.returncode = 0
        return proc

    supervisor = Supervisor(
        cfg, with_tui=False, overrides=overrides, spawn=fake_spawn, poll_s=0.001
    )
    supervisor.run()

    daemon_argv = spawned[0]
    assert "--device" in daemon_argv and "microphone.usb" in daemon_argv
    assert "--backend" in daemon_argv and "sounddevice" in daemon_argv
    assert "--model" in daemon_argv and "base" in daemon_argv
    assert "--language" in daemon_argv and "es" in daemon_argv
    assert "--config" in daemon_argv and "/path/to/config.toml" in daemon_argv


def test_supervisor_forwards_verbose():
    """Verbose flag forwarded to daemon and assistant."""
    cfg = _cfg()
    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        spawned.append(argv)
        proc = FakePopen(argv, redirect_output=redirect_output)
        if len(spawned) == 1:
            proc.returncode = 0
        return proc

    supervisor = Supervisor(cfg, with_tui=False, verbose=True, spawn=fake_spawn, poll_s=0.001)
    supervisor.run()

    assert "-v" in spawned[0]
    assert "-v" in spawned[1]


def test_supervisor_skips_config_when_not_given():
    """--config flag omitted when not in overrides."""
    cfg = _cfg()
    overrides = {"device": "test"}
    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        spawned.append(argv)
        proc = FakePopen(argv, redirect_output=redirect_output)
        if len(spawned) == 1:
            proc.returncode = 0
        return proc

    supervisor = Supervisor(
        cfg, with_tui=False, overrides=overrides, spawn=fake_spawn, poll_s=0.001
    )
    supervisor.run()

    assert "--config" not in spawned[0]
    assert "--device" in spawned[0]


def test_supervisor_assistant_receives_only_allowed_flags():
    """Assistant gets config and verbose, but not device/backend/model."""
    cfg = _cfg()
    overrides = {
        "config": "/config.toml",
        "device": "ignore",
        "backend": "ignore",
        "model": "ignore",
    }
    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        spawned.append(argv)
        proc = FakePopen(argv, redirect_output=redirect_output)
        if len(spawned) == 1:
            proc.returncode = 0
        return proc

    supervisor = Supervisor(
        cfg, with_tui=False, overrides=overrides, verbose=True, spawn=fake_spawn, poll_s=0.001
    )
    supervisor.run()

    assistant_argv = spawned[1]
    assert "--config" in assistant_argv
    assert "/config.toml" in assistant_argv
    assert "-v" in assistant_argv
    assert "--device" not in assistant_argv
    assert "--backend" not in assistant_argv
    assert "--model" not in assistant_argv


def test_supervisor_spawn_redirect_output_false_without_tui():
    """spawn() is called with redirect_output=False when with_tui=False."""
    cfg = _cfg()
    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        spawned.append((argv, redirect_output))
        proc = FakePopen(argv, redirect_output=redirect_output)
        if len(spawned) == 1:
            proc.returncode = 0
        return proc

    supervisor = Supervisor(cfg, with_tui=False, spawn=fake_spawn, poll_s=0.001)
    supervisor.run()

    for argv, redirect_output in spawned:
        assert redirect_output is False


def test_supervisor_spawn_redirect_output_true_with_tui():
    """spawn() is called with redirect_output=True when with_tui=True."""
    cfg = _cfg()
    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        spawned.append((argv, redirect_output))
        proc = FakePopen(argv, redirect_output=redirect_output)
        if len(spawned) == 1:
            proc.returncode = 0
        return proc

    supervisor = Supervisor(cfg, with_tui=True, spawn=fake_spawn, wait_timeout_s=0.01, poll_s=0.001)
    supervisor.run()

    for argv, redirect_output in spawned:
        assert redirect_output is True


def test_supervisor_run_spawns_tui_after_control_socket(tmp_path):
    """With_tui=True spawns the TUI once the control socket exists; TUI exit stops all."""
    cfg = _cfg()
    cfg.assistant.control_socket = str(tmp_path / "assistant.sock")
    (tmp_path / "assistant.sock").touch()

    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        proc = FakePopen(argv, redirect_output=redirect_output)
        spawned.append((argv, proc))
        if len(spawned) == 3:
            proc.returncode = 0
        return proc

    clock_val = [0.0]

    def fake_clock():
        return clock_val[0]

    def fake_sleep(_s):
        clock_val[0] += 0.1

    supervisor = Supervisor(
        cfg,
        with_tui=True,
        spawn=fake_spawn,
        sleep=fake_sleep,
        clock=fake_clock,
        poll_s=0.001,
    )
    result = supervisor.run()

    assert len(spawned) == 3
    assert "stt_daemon.tui" in " ".join(spawned[2][0])
    assert spawned[0][1].redirect_output is True
    assert spawned[1][1].redirect_output is True
    assert spawned[2][1].redirect_output is False
    assert result == 0
    for _argv, proc in spawned[:2]:
        assert proc._terminated or proc._killed


def test_supervisor_force_kills_stubborn_children():
    """A child that ignores SIGTERM is force-killed after the grace period."""
    cfg = _cfg()

    class StubbornPopen(FakePopen):
        def terminate(self):
            self._terminated = True

    daemon = StubbornPopen(["daemon"])

    clock_val = [0.0]

    def fake_clock():
        return clock_val[0]

    def fake_sleep(_s):
        clock_val[0] += 0.1

    supervisor = Supervisor(
        cfg, with_tui=False, grace_s=0.05, clock=fake_clock, sleep=fake_sleep
    )
    supervisor._terminate_all([daemon])

    assert daemon._killed


def test_supervisor_run_signal_shutdown_returns_zero(monkeypatch):
    """SIGINT stops the loop, terminates every child, and returns 0."""
    cfg = _cfg()
    handlers = {}
    monkeypatch.setattr(
        "stt_daemon.supervisor.signal.signal",
        lambda signum, handler: handlers.__setitem__(signum, handler),
    )

    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        proc = FakePopen(argv, redirect_output=redirect_output)
        spawned.append(proc)
        return proc

    clock_val = [0.0]

    def fake_clock():
        return clock_val[0]

    def fake_sleep(_s):
        clock_val[0] += 0.1
        if signal.SIGINT in handlers:
            handlers[signal.SIGINT](signal.SIGINT, None)

    supervisor = Supervisor(
        cfg,
        with_tui=False,
        spawn=fake_spawn,
        sleep=fake_sleep,
        clock=fake_clock,
        poll_s=0.001,
    )
    result = supervisor.run()

    assert len(spawned) == 2
    assert result == 0
    for proc in spawned:
        assert proc._terminated or proc._killed


def test_supervisor_restores_terminal_after_tui(monkeypatch, tmp_path):
    """A killed TUI must not leave the terminal in raw mode."""
    cfg = _cfg()
    cfg.assistant.control_socket = str(tmp_path / "assistant.sock")
    (tmp_path / "assistant.sock").touch()

    log = []
    monkeypatch.setattr("stt_daemon.supervisor.termios", FakeTermios(log))
    monkeypatch.setattr(
        "stt_daemon.supervisor.sys",
        types.SimpleNamespace(stdin=FakeStdin(), executable=sys.executable),
    )

    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        proc = FakePopen(argv, redirect_output=redirect_output)
        spawned.append(proc)
        if len(spawned) == 3:
            proc.returncode = 0
        return proc

    clock_val = [0.0]

    def fake_clock():
        return clock_val[0]

    def fake_sleep(_s):
        clock_val[0] += 0.1

    supervisor = Supervisor(
        cfg,
        with_tui=True,
        spawn=fake_spawn,
        sleep=fake_sleep,
        clock=fake_clock,
        poll_s=0.001,
    )
    supervisor.run()

    assert ("get", 7) in log
    assert ("set", 7, FakeTermios.TCSADRAIN, [0, 0, 0, 1, 0, 0, []]) in log


def test_supervisor_leaves_terminal_alone_without_tui(monkeypatch):
    """Without --with-tui the supervisor never touches terminal attributes."""
    cfg = _cfg()
    log = []
    monkeypatch.setattr("stt_daemon.supervisor.termios", FakeTermios(log))
    monkeypatch.setattr(
        "stt_daemon.supervisor.sys",
        types.SimpleNamespace(stdin=FakeStdin(), executable=sys.executable),
    )

    spawned = []

    def fake_spawn(argv, *, redirect_output=False):
        proc = FakePopen(argv, redirect_output=redirect_output)
        spawned.append(proc)
        if len(spawned) == 1:
            proc.returncode = 0
        return proc

    supervisor = Supervisor(cfg, with_tui=False, spawn=fake_spawn, poll_s=0.001)
    supervisor.run()

    assert log == []