led-controller/tests/test_audio_device_select.py

"""Audio input device_select persistence (Pulse name must survive start)."""

import asyncio
import json
import os
import sys
import threading
import time
from pathlib import Path
from unittest.mock import MagicMock

import pytest
import requests

PROJECT_ROOT = Path(__file__).resolve().parents[1]
SRC_PATH = PROJECT_ROOT / "src"
if str(SRC_PATH) not in sys.path:
    sys.path.insert(0, str(SRC_PATH))

from microdot import Microdot  # noqa: E402
from util.audio_run_persist import read_audio_run_state, write_audio_run_state  # noqa: E402

SNOWBALL = (
    "alsa_input.usb-BLUE_MICROPHONE_Blue_Snowball_SUGA_2020_10_09_41646-00.mono-fallback"
)


def _start_app(app: Microdot, port: int = 0):
    def runner():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            loop.run_until_complete(app.start_server(host="127.0.0.1", port=port))
        finally:
            loop.close()

    thread = threading.Thread(target=runner, daemon=True)
    thread.start()
    deadline = time.time() + 5.0
    while time.time() < deadline:
        server = getattr(app, "server", None)
        if server and getattr(server, "sockets", None):
            sockets = server.sockets or []
            if sockets:
                return thread, sockets[0].getsockname()[1]
        time.sleep(0.05)
    raise RuntimeError("server failed to start")


@pytest.fixture
def audio_run_path(tmp_path, monkeypatch):
    path = tmp_path / "audio_run.json"
    monkeypatch.setattr(
        "util.audio_run_persist._db_path",
        lambda: str(path),
    )
    return path


def test_write_start_keeps_pulse_device_select_not_portaudio_index(audio_run_path):
    write_audio_run_state(
        enabled=True,
        device=1,
        device_select=SNOWBALL,
    )
    st = read_audio_run_state()
    assert st["device_select"] == SNOWBALL
    assert st["device"] == 1


def test_put_device_saves_pulse_name(audio_run_path):
    app = Microdot()

    @app.route("/api/audio/device", methods=["PUT"])
    async def audio_set_device(request):
        payload = request.json if isinstance(request.json, dict) else {}
        device_select = str(payload.get("device_select") or "").strip()
        from util.audio_run_persist import read_audio_run_state, write_audio_run_state

        prev = read_audio_run_state()
        write_audio_run_state(
            enabled=bool(prev.get("enabled")),
            device=device_select if device_select else None,
            device_override="",
            device_select=device_select,
        )
        return {"ok": True, "audio_run": read_audio_run_state()}

    _, port = _start_app(app)
    base = f"http://127.0.0.1:{port}"
    resp = requests.put(
        f"{base}/api/audio/device",
        json={"device_select": SNOWBALL, "device_override": ""},
        timeout=5,
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["audio_run"]["device_select"] == SNOWBALL
    assert read_audio_run_state()["device_select"] == SNOWBALL


def test_start_preserves_device_select_in_status(audio_run_path, monkeypatch):
    detector = MagicMock()
    detector.status.return_value = {"running": True, "device": 2}

    def fake_resolve(device):
        assert device == SNOWBALL
        return 2

    monkeypatch.setattr(
        "util.pulse_audio_devices.resolve_capture_device",
        fake_resolve,
    )

    app = Microdot()

    @app.route("/api/audio/start", methods=["POST"])
    async def audio_start(request):
        payload = request.json if isinstance(request.json, dict) else {}
        device = payload.get("device", None)
        if device in ("", None):
            device = None
        device_select = str(payload.get("device_select") or "").strip()
        if not device_select and device not in ("", None):
            device_select = str(device).strip()
        from util.pulse_audio_devices import resolve_capture_device
        from util.audio_run_persist import read_audio_run_state, write_audio_run_state

        device = resolve_capture_device(device)
        detector.start(device=device)
        write_audio_run_state(
            enabled=True,
            device=device,
            device_override="",
            device_select=device_select,
        )
        st = detector.status()
        st["audio_run"] = read_audio_run_state()
        return {"ok": True, "status": st}

    @app.route("/api/audio/status")
    async def audio_status(request):
        _ = request
        from util.audio_run_persist import read_audio_run_state

        st = detector.status()
        st["audio_run"] = read_audio_run_state()
        return {"status": st}

    _, port = _start_app(app)
    base = f"http://127.0.0.1:{port}"
    start = requests.post(
        f"{base}/api/audio/start",
        json={"device": SNOWBALL, "device_select": SNOWBALL, "device_override": ""},
        timeout=5,
    )
    assert start.status_code == 200, start.text
    run = start.json()["status"]["audio_run"]
    assert run["device_select"] == SNOWBALL
    assert run["device"] == 2

    status = requests.get(f"{base}/api/audio/status", timeout=5).json()["status"]
    assert status["audio_run"]["device_select"] == SNOWBALL


def test_pulse_device_list_uses_stable_pulse_ids():
    from util.pulse_audio_devices import list_pulse_matched_input_devices

    devs = list_pulse_matched_input_devices()
    if not devs:
        pytest.skip("pactl not available")
    snow = next((d for d in devs if "Snowball" in d.get("display_name", "")), None)
    if snow is None:
        pytest.skip("Blue Snowball not connected")
    assert snow["id"] == snow["pulse_name"]
    assert str(snow["id"]).startswith("alsa_input.")