STT-Assistant/tests/test_pulse.py

"""Tests for PulseAudio interface with injectable runner.

Fixtures mirror the ACTUAL shape of `pactl -f json list sources|sinks`:
a top-level JSON array, where monitor sources carry a non-empty
``monitor_source`` (the sink they mirror) and real inputs carry ``""``.
"""

import json

from stt_daemon.pulse import Device, PulseAudio

REAL_SOURCES = [
    {
        "index": 12,
        "name": "alsa_output.pci-0000_05_00.1.hdmi-stereo.monitor",
        "description": "Monitor of Radeon HDMI",
        "monitor_source": "alsa_output.pci-0000_05_00.1.hdmi-stereo",
    },
    {
        "index": 15,
        "name": "alsa_input.usb-C-Media.mono-fallback",
        "description": "USB Audio Mono",
        "monitor_source": "",
    },
]

REAL_SINKS = [
    {
        "index": 12,
        "name": "alsa_output.pci-0000_05_00.1.hdmi-stereo",
        "description": "Radeon HDMI",
        "monitor_source": "alsa_output.pci-0000_05_00.1.hdmi-stereo.monitor",
    },
    {
        "index": 13,
        "name": "alsa_output.usb-analog-stereo",
        "description": "USB Analog",
        "monitor_source": "alsa_output.usb-analog-stereo.monitor",
    },
]


def _make_pa(sources=None, sinks=None):
    def fake_runner(args):
        if "list" in args and "sources" in args:
            return json.dumps(sources if sources is not None else [])
        if "list" in args and "sinks" in args:
            return json.dumps(sinks if sinks is not None else [])
        raise RuntimeError(f"Unexpected args: {args}")

    return PulseAudio(runner=fake_runner)


def test_list_devices_excludes_monitors_and_uses_sink_monitor_source():
    """A .monitor source must NOT be an input; each sink contributes its monitor as output."""
    pa = _make_pa(REAL_SOURCES, REAL_SINKS)
    devices = pa.list_devices()

    inputs = [d for d in devices if d.kind == "input"]
    outputs = [d for d in devices if d.kind == "output"]

    assert [d.name for d in inputs] == ["alsa_input.usb-C-Media.mono-fallback"]
    assert [d.name for d in outputs] == [
        "alsa_output.pci-0000_05_00.1.hdmi-stereo.monitor",
        "alsa_output.usb-analog-stereo.monitor",
    ]
    assert outputs[0].description == "Radeon HDMI"


def test_list_devices_exclude():
    """Excluded names (our own virtual devices) are omitted."""
    pa = _make_pa(REAL_SOURCES, REAL_SINKS)
    devices = pa.list_devices(
        exclude={
            "alsa_input.usb-C-Media.mono-fallback",
            "alsa_output.usb-analog-stereo.monitor",
        }
    )
    assert [d.name for d in devices] == [
        "alsa_output.pci-0000_05_00.1.hdmi-stereo.monitor"
    ]


def test_list_sources_tolerates_dict_wrapped_output():
    """Parser must not crash on the legacy object-wrapped form."""
    def runner(args):
        return json.dumps(
            {"sources": [{"name": "in1", "description": "In 1", "monitor_source": ""}]}
        )

    pa = PulseAudio(runner=runner)
    assert [s["name"] for s in pa.list_sources()] == ["in1"]


def test_sink_exists():
    """Test sink existence check."""
    pa = _make_pa(REAL_SOURCES, REAL_SINKS)

    assert pa.sink_exists("alsa_output.usb-analog-stereo")
    assert not pa.sink_exists("VirtualMicSink")


def test_source_exists():
    """Test source existence check."""
    pa = _make_pa(REAL_SOURCES, REAL_SINKS)

    assert pa.source_exists("alsa_input.usb-C-Media.mono-fallback")
    assert pa.source_exists("alsa_output.pci-0000_05_00.1.hdmi-stereo.monitor")
    assert not pa.source_exists("nonexistent")


def test_load_module_parses_id():
    """Test that load_module parses the numeric ID from pactl output."""
    def fake_runner(args):
        if "load-module" in args:
            return "42\n"
        raise RuntimeError(f"Unexpected args: {args}")

    pa = PulseAudio(runner=fake_runner)
    assert pa.load_module("module-null-sink", "sink_name=VirtualMicSink") == 42


def test_unload_module_success():
    """Test unload_module returns True on success."""
    calls = []

    def fake_runner(args):
        if "unload-module" in args:
            calls.append(args)
            return ""
        raise RuntimeError(f"Unexpected args: {args}")

    pa = PulseAudio(runner=fake_runner)
    assert pa.unload_module(42) is True
    assert len(calls) == 1


def test_unload_module_already_gone_real_message():
    """pactl reports a vanished module as 'Failure: No such entity'."""
    def fake_runner(args):
        if "unload-module" in args:
            raise RuntimeError("pactl failed: Failure: No such entity\n")
        raise RuntimeError(f"Unexpected args: {args}")

    pa = PulseAudio(runner=fake_runner)
    assert pa.unload_module(42) is False


def test_short_modules_parsing():
    """Test that short_modules parses pactl list short modules output."""
    output = "0\tmodule-cli-protocol-tcp\n12\tmodule-null-sink\n13\tmodule-loopback\n"

    def fake_runner(args):
        if "short" in args and "modules" in args:
            return output
        raise RuntimeError(f"Unexpected args: {args}")

    pa = PulseAudio(runner=fake_runner)
    modules = pa.short_modules()

    assert modules == [
        (0, "module-cli-protocol-tcp"),
        (12, "module-null-sink"),
        (13, "module-loopback"),
    ]


def test_default_sink_name():
    """Test default_sink_name returns the sink name."""
    def fake_runner(args):
        if "get-default-sink" in args:
            return "alsa_output.usb-analog-stereo\n"
        raise RuntimeError(f"Unexpected args: {args}")

    pa = PulseAudio(runner=fake_runner)
    assert pa.default_sink_name() == "alsa_output.usb-analog-stereo"


def test_device_dataclass():
    """Test Device dataclass."""
    device = Device(name="alsa_input.usb-Foo", description="USB Input", kind="input")

    assert device.name == "alsa_input.usb-Foo"
    assert device.description == "USB Input"
    assert device.kind == "input"