led-controller/tests/test_audio_device_select.py

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

import sys
from pathlib import Path
from unittest.mock import MagicMock

import pytest
from fastapi import FastAPI
from starlette.testclient import TestClient

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 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"
)


@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):
    api = FastAPI()

    @api.put("/api/audio/device")
    async def audio_set_device(payload: dict | None = None):
        body = payload if isinstance(payload, dict) else {}
        device_select = str(body.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()}

    with TestClient(api) as client:
        resp = client.put(
            "/api/audio/device",
            json={"device_select": SNOWBALL, "device_override": ""},
        )
    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,
    )

    api = FastAPI()

    @api.post("/api/audio/start")
    async def audio_start(payload: dict | None = None):
        body = payload if isinstance(payload, dict) else {}
        device = body.get("device", None)
        if device in ("", None):
            device = None
        device_select = str(body.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}

    @api.get("/api/audio/status")
    async def audio_status():
        from util.audio_run_persist import read_audio_run_state

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

    with TestClient(api) as client:
        start = client.post(
            "/api/audio/start",
            json={"device": SNOWBALL, "device_select": SNOWBALL, "device_override": ""},
        )
        assert start.status_code == 200, start.text
        run = start.json()["status"]["audio_run"]
        assert run["device_select"] == SNOWBALL
        assert run["device"] == 2

        status = client.get("/api/audio/status").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.")