STT-Assistant/tests/test_provisioner.py

"""Tests for audio sink provisioner orchestration."""


import pytest

from stt_daemon.config import SinkConfig
from stt_daemon.provisioner import Provisioner
from stt_daemon.pulse import PulseAudio
from stt_daemon.state import AudioState, Loopback, StateStore


def test_provisioner_setup_creates_modules_in_order(tmp_path):
    """Test that setup creates modules in the correct order."""
    state_file = tmp_path / "audio.json"
    store = StateStore(state_file)
    cfg = SinkConfig(sink_name="VirtualMicSink", remap_source_name="VirtualMic")

    call_sequence = []

    def fake_runner(args):
        if "load-module" in args:
            module_id = len(call_sequence) + 10
            call_sequence.append((args[1], module_id))
            return f"{module_id}\n"
        elif "list" in args and "sinks" in args:
            return '{"sinks": []}'
        elif "list" in args and "sources" in args:
            return '{"sources": []}'
        else:
            return ""

    pa = PulseAudio(runner=fake_runner)
    provisioner = Provisioner(pa, cfg, store)

    provisioner.setup(["alsa_input.usb-Foo", "alsa_output.pci-0000.monitor"])

    # Verify creation order: null-sink, remap, then loopbacks
    assert call_sequence[0][0] == "module-null-sink"
    assert call_sequence[1][0] == "module-remap-source"
    assert call_sequence[2][0] == "module-loopback"
    assert call_sequence[3][0] == "module-loopback"

    # Verify state is saved
    loaded_state = store.load()
    assert loaded_state is not None
    assert loaded_state.sink_name == "VirtualMicSink"
    assert len(loaded_state.loopbacks) == 2


def test_provisioner_teardown_unloads_in_reverse_order(tmp_path):
    """Test that teardown unloads modules in reverse order."""
    state_file = tmp_path / "audio.json"
    store = StateStore(state_file)
    cfg = SinkConfig(sink_name="VirtualMicSink", remap_source_name="VirtualMic")

    # Pre-save a state
    state = AudioState(
        sink_name="VirtualMicSink",
        remap_name="VirtualMic",
        null_sink_module=10,
        remap_module=11,
        loopbacks=[
            Loopback(module_id=12, source="alsa_input.usb-Foo"),
            Loopback(module_id=13, source="alsa_output.pci-0000.monitor"),
        ],
    )
    store.save(state)

    unload_sequence = []

    def fake_runner(args):
        if "unload-module" in args:
            module_id = int(args[1])
            unload_sequence.append(module_id)
            return ""
        elif "list" in args:
            return '{"sources": [], "sinks": []}'
        else:
            return ""

    pa = PulseAudio(runner=fake_runner)
    provisioner = Provisioner(pa, cfg, store)

    provisioner.teardown()

    # Verify unload order: loopbacks in reverse, then remap, then null-sink
    assert unload_sequence == [13, 12, 11, 10]

    # Verify state is cleared
    assert store.load() is None


def test_provisioner_ensure_returns_none_if_no_state(tmp_path):
    """Test that ensure returns None if no saved state exists."""
    state_file = tmp_path / "audio.json"
    store = StateStore(state_file)
    cfg = SinkConfig(sink_name="VirtualMicSink", remap_source_name="VirtualMic")

    def fake_runner(args):
        if "list" in args and "sinks" in args:
            return '{"sinks": []}'
        else:
            return '{"sources": []}'

    pa = PulseAudio(runner=fake_runner)
    provisioner = Provisioner(pa, cfg, store)

    result = provisioner.ensure()

    assert result is None


def test_provisioner_ensure_returns_state_if_sink_exists(tmp_path):
    """Test that ensure returns existing state if sink is present."""
    state_file = tmp_path / "audio.json"
    store = StateStore(state_file)
    cfg = SinkConfig(sink_name="VirtualMicSink", remap_source_name="VirtualMic")

    # Pre-save a state
    state = AudioState(
        sink_name="VirtualMicSink",
        remap_name="VirtualMic",
        null_sink_module=10,
        remap_module=11,
        loopbacks=[],
    )
    store.save(state)

    def fake_runner(args):
        if "list" in args and "sinks" in args:
            return (
                '{"sinks": [{"name": "VirtualMicSink", '
                '"monitor_source": "VirtualMicSink.monitor"}]}'
            )
        else:
            return '{"sources": []}'

    pa = PulseAudio(runner=fake_runner)
    provisioner = Provisioner(pa, cfg, store)

    result = provisioner.ensure()

    assert result is not None
    assert result.sink_name == "VirtualMicSink"


def test_provisioner_ensure_reprovisions_if_sink_missing(tmp_path):
    """Test that ensure re-provisions if saved state exists but sink is gone."""
    state_file = tmp_path / "audio.json"
    store = StateStore(state_file)
    cfg = SinkConfig(sink_name="VirtualMicSink", remap_source_name="VirtualMic")

    # Pre-save a state
    original_state = AudioState(
        sink_name="VirtualMicSink",
        remap_name="VirtualMic",
        null_sink_module=10,
        remap_module=11,
        loopbacks=[Loopback(module_id=12, source="alsa_input.usb-Foo")],
        selected_sources=["alsa_input.usb-Foo"],
    )
    store.save(original_state)

    module_counter = [20]

    def fake_runner(args):
        if "list" in args and "sinks" in args:
            # Sink doesn't exist
            return '{"sinks": []}'
        elif "list" in args and "sources" in args:
            return '{"sources": []}'
        elif "load-module" in args:
            module_id = module_counter[0]
            module_counter[0] += 1
            return f"{module_id}\n"
        else:
            return ""

    pa = PulseAudio(runner=fake_runner)
    provisioner = Provisioner(pa, cfg, store)

    result = provisioner.ensure()

    assert result is not None
    assert result.sink_name == "VirtualMicSink"
    assert result.selected_sources == ["alsa_input.usb-Foo"]
    # Should have created new modules
    assert result.null_sink_module == 20


def test_provisioner_setup_raises_on_untracked_sink(tmp_path):
    """Test that setup raises if sink already exists but is not tracked."""
    state_file = tmp_path / "audio.json"
    store = StateStore(state_file)
    cfg = SinkConfig(sink_name="VirtualMicSink", remap_source_name="VirtualMic")

    def fake_runner(args):
        if "list" in args and "sinks" in args:
            # VirtualMicSink already exists (from elsewhere)
            return (
                '{"sinks": [{"name": "VirtualMicSink", '
                '"monitor_source": "VirtualMicSink.monitor"}]}'
            )
        else:
            return '{"sources": []}'

    pa = PulseAudio(runner=fake_runner)
    provisioner = Provisioner(pa, cfg, store)

    with pytest.raises(RuntimeError, match="already exists but is not tracked"):
        provisioner.setup(["alsa_input.usb-Foo"])


def test_provisioner_status_reports_live_modules(tmp_path):
    """Test that status reports which tracked modules are currently loaded."""
    state_file = tmp_path / "audio.json"
    store = StateStore(state_file)
    cfg = SinkConfig(sink_name="VirtualMicSink", remap_source_name="VirtualMic")

    state = AudioState(
        sink_name="VirtualMicSink",
        remap_name="VirtualMic",
        null_sink_module=10,
        remap_module=11,
        loopbacks=[Loopback(module_id=12, source="alsa_input.usb-Foo")],
    )
    store.save(state)

    def fake_runner(args):
        if "list" in args and "sinks" in args:
            return '{"sinks": [{"name": "VirtualMicSink"}]}'
        elif "list" in args and "short" in args and "modules" in args:
            # Only 2 of 3 tracked modules are loaded
            return "10\tmodule-null-sink\n11\tmodule-remap-source\n"
        else:
            return '{"sources": []}'

    pa = PulseAudio(runner=fake_runner)
    provisioner = Provisioner(pa, cfg, store)

    status = provisioner.status()

    assert status["state"] is not None
    assert status["sink_present"] is True
    assert len(status["live_modules"]) == 2


def test_provisioner_teardown_tolerates_missing_modules(tmp_path):
    """Test that teardown tolerates modules that are already gone."""
    state_file = tmp_path / "audio.json"
    store = StateStore(state_file)
    cfg = SinkConfig(sink_name="VirtualMicSink", remap_source_name="VirtualMic")

    state = AudioState(
        sink_name="VirtualMicSink",
        remap_name="VirtualMic",
        null_sink_module=10,
        remap_module=11,
    )
    store.save(state)

    unload_attempts = []

    def fake_runner(args):
        if "unload-module" in args:
            module_id = int(args[1])
            unload_attempts.append(module_id)
            # Simulate module 11 already being gone
            if module_id == 11:
                raise RuntimeError("Module 11 not found")
            return ""
        else:
            return '{"sources": [], "sinks": []}'

    pa = PulseAudio(runner=fake_runner)
    provisioner = Provisioner(pa, cfg, store)

    # Should not raise
    provisioner.teardown()

    # State should still be cleared
    assert store.load() is None