musicmouse/python-backend/tests/test_screen.py

"""The screen service: runs the configured command per toggle, and never takes the bus down."""

from __future__ import annotations

import asyncio
import io
import os
import shlex
import sys
import wave
from collections.abc import AsyncIterator
from pathlib import Path

import pytest

from musicmouse.blubb import RATE, blubb_wav
from musicmouse.bus import EventBus
from musicmouse.clock import FakeClock
from musicmouse.color import ColorRGBW
from musicmouse.config import ScreenConfig
from musicmouse.effects import EffectStaticConfig
from musicmouse.events import (
    LedEffectRequested,
    PlaybackChanged,
    PlayRemoteSlotRequested,
    ToggleScreenRequested,
    UserActivity,
)
from musicmouse.hardware import LedZone
from musicmouse.services.screen import ScreenService


def _python(code: str) -> str:
    return f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}"


def _appending(log: Path, *, then: str = "") -> str:
    """A command that appends a line to ``log`` each time it runs."""
    return _python(f"open({str(log)!r}, 'a').write('ran\\n'); {then}")


@pytest.fixture
async def bus() -> AsyncIterator[EventBus]:
    async with EventBus() as bus:
        yield bus


async def _settle(service: ScreenService) -> None:
    while service._running is not None and not service._running.done():
        await asyncio.sleep(0.01)


async def _started(
    bus: EventBus,
    config: ScreenConfig,
    *,
    alsa_device: str | None = None,
    volume: float = 0.5,
) -> tuple[ScreenService, asyncio.Task[None]]:
    service = ScreenService(bus, config, alsa_device=alsa_device, volume=lambda: volume)
    task = asyncio.create_task(service.run())
    await asyncio.sleep(0)  # let it subscribe
    return service, task


async def test_each_press_runs_the_command(bus: EventBus, tmp_path: Path) -> None:
    log = tmp_path / "ran"
    service, task = await _started(bus, ScreenConfig(toggle_command=_appending(log)))
    try:
        for _ in range(2):
            await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
            await _settle(service)
    finally:
        task.cancel()

    assert log.read_text() == "ran\nran\n"


async def test_a_press_while_the_command_runs_is_ignored(bus: EventBus, tmp_path: Path) -> None:
    log = tmp_path / "ran"
    command = _appending(log, then="import time; time.sleep(0.3)")
    service, task = await _started(bus, ScreenConfig(toggle_command=command))
    try:
        await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
        await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
        await _settle(service)
    finally:
        task.cancel()

    assert log.read_text() == "ran\n"


async def test_a_failing_command_is_logged(bus: EventBus, caplog: pytest.LogCaptureFixture) -> None:
    command = _python("import sys; print('no display'); sys.exit(3)")
    service, task = await _started(bus, ScreenConfig(toggle_command=command))
    try:
        await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
        await _settle(service)
    finally:
        task.cancel()

    assert "exit 3" in caplog.text
    assert "no display" in caplog.text


async def test_a_missing_command_is_logged(
    bus: EventBus, caplog: pytest.LogCaptureFixture, tmp_path: Path
) -> None:
    service, task = await _started(
        bus, ScreenConfig(toggle_command=str(tmp_path / "no-such-command"))
    )
    try:
        await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
        await _settle(service)
    finally:
        task.cancel()

    assert "could not start" in caplog.text


async def test_a_hung_command_is_killed(bus: EventBus, caplog: pytest.LogCaptureFixture) -> None:
    command = _python("import time; time.sleep(30)")
    service, task = await _started(bus, ScreenConfig(toggle_command=command, timeout=0.2))
    try:
        await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
        await asyncio.wait_for(_settle(service), 5)
    finally:
        task.cancel()

    assert "killed" in caplog.text


# ------------------------------------------------------------------------ the blubb


def test_the_blubb_is_a_short_mono_wav() -> None:
    with wave.open(io.BytesIO(blubb_wav(on=True, gain=0.5))) as wav:
        assert (wav.getnchannels(), wav.getsampwidth(), wav.getframerate()) == (1, 2, RATE)
        assert 0.1 < wav.getnframes() / RATE < 0.5


def test_on_and_off_sound_different() -> None:
    assert blubb_wav(on=True, gain=0.5) != blubb_wav(on=False, gain=0.5)


@pytest.fixture
def fake_aplay(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    """An ``aplay`` on PATH that records its arguments and what it was fed."""
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    record = tmp_path / "aplay"
    script = bin_dir / "aplay"
    script.write_text(
        f'#!/bin/sh\necho "$@" > {shlex.quote(str(record))}.args\n'
        f"cat > {shlex.quote(str(record))}.wav\n"
    )
    script.chmod(0o755)
    monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}")
    return record


async def test_a_switch_plays_the_blubb_its_way(bus: EventBus, fake_aplay: Path) -> None:
    service, task = await _started(
        bus, ScreenConfig(toggle_command=_python("print('off')")), alsa_device="softvol_effects"
    )
    try:
        await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
        await _settle(service)
    finally:
        task.cancel()

    assert Path(f"{fake_aplay}.args").read_text().split() == ["-q", "-D", "softvol_effects", "-"]
    assert Path(f"{fake_aplay}.wav").read_bytes() == blubb_wav(on=False, gain=0.35 * 0.5)


@pytest.mark.parametrize(
    ("output", "alsa_device", "volume"),
    [
        ("print('on')", None, 0.5),  # simulated audio
        ("print('on')", "softvol_effects", 0.0),  # muted
        ("print('done')", "softvol_effects", 0.5),  # no direction to go by
    ],
)
async def test_no_blubb(
    bus: EventBus, fake_aplay: Path, output: str, alsa_device: str | None, volume: float
) -> None:
    service, task = await _started(
        bus, ScreenConfig(toggle_command=_python(output)), alsa_device=alsa_device, volume=volume
    )
    try:
        await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
        await _settle(service)
    finally:
        task.cancel()

    assert not Path(f"{fake_aplay}.wav").exists()


# ------------------------------------------------------------------- idle auto-off


class _Idle:
    """A started service with a fake clock and an ``off`` and ``on`` command that log."""

    def __init__(self, bus: EventBus, tmp_path: Path, **config: object) -> None:
        self.bus = bus
        self.off_log = tmp_path / "off"
        self.on_log = tmp_path / "on"
        self.playing = False
        self.clock = FakeClock(idle=bus.drain)
        settings: dict[str, object] = {
            "toggle_command": _python("print('toggled')"),
            "off_command": _appending(self.off_log, then="print('off')"),
            "on_command": _appending(self.on_log, then="print('on')"),
            "auto_off_minutes": 15,
        }
        settings.update(config)
        self.service = ScreenService(
            bus,
            ScreenConfig.model_validate(settings),
            alsa_device=None,
            volume=lambda: 0.5,
            is_playing=lambda: self.playing,
            clock=self.clock,
        )
        self.task: asyncio.Task[None] | None = None

    async def __aenter__(self) -> _Idle:
        self.task = asyncio.create_task(self.service.run())
        await asyncio.sleep(0)
        return self

    async def __aexit__(self, *_: object) -> None:
        assert self.task is not None
        self.task.cancel()

    async def wait(self, minutes: float) -> None:
        await self.clock.advance(minutes * 60)
        await _settle(self.service)

    def offs(self) -> int:
        return len(self.off_log.read_text().splitlines()) if self.off_log.exists() else 0

    def ons(self) -> int:
        return len(self.on_log.read_text().splitlines()) if self.on_log.exists() else 0


async def test_the_screen_goes_off_after_the_idle_time(bus: EventBus, tmp_path: Path) -> None:
    async with _Idle(bus, tmp_path) as idle:
        await idle.wait(14)
        assert idle.offs() == 0
        await idle.wait(1)
        assert idle.offs() == 1
        await idle.wait(60)
        assert idle.offs() == 1  # once, not every half minute after


async def test_not_while_music_plays(bus: EventBus, tmp_path: Path) -> None:
    async with _Idle(bus, tmp_path) as idle:
        idle.playing = True
        await idle.wait(60)
        assert idle.offs() == 0

        # The countdown starts when the music stops.
        idle.playing = False
        await bus.emit_and_wait(PlaybackChanged(playing=False, source="player"))
        await idle.wait(14)
        assert idle.offs() == 0
        await idle.wait(1)
        assert idle.offs() == 1


@pytest.mark.parametrize(
    "event",
    [UserActivity(source="web"), PlayRemoteSlotRequested(digit="3", source="lirc")],
)
async def test_activity_restarts_the_countdown(
    bus: EventBus, tmp_path: Path, event: UserActivity | PlayRemoteSlotRequested
) -> None:
    async with _Idle(bus, tmp_path) as idle:
        await idle.wait(10)
        await bus.emit_and_wait(event)
        await idle.wait(10)
        assert idle.offs() == 0
        await idle.wait(5)
        assert idle.offs() == 1


async def test_automations_do_not_count_as_activity(bus: EventBus, tmp_path: Path) -> None:
    async with _Idle(bus, tmp_path) as idle:
        await idle.wait(10)
        effect = EffectStaticConfig(ColorRGBW(0, 0, 0, 0))
        automation = LedEffectRequested(zone=LedZone.MOUSE, effect=effect, source="mqtt")
        await bus.emit_and_wait(automation)
        await idle.wait(5)
        assert idle.offs() == 1


async def test_someone_coming_back_switches_it_on(bus: EventBus, tmp_path: Path) -> None:
    async with _Idle(bus, tmp_path) as idle:
        await idle.wait(15)
        assert idle.offs() == 1

        await bus.emit_and_wait(UserActivity(source="web"))
        await _settle(idle.service)
        await bus.emit_and_wait(UserActivity(source="web"))
        await _settle(idle.service)
        assert idle.ons() == 1  # only the first wakes it

        await idle.wait(15)
        assert idle.offs() == 2  # and the countdown runs again


async def test_the_power_key_after_an_auto_off_is_left_to_the_toggle(
    bus: EventBus, tmp_path: Path
) -> None:
    async with _Idle(bus, tmp_path) as idle:
        await idle.wait(15)
        await bus.emit_and_wait(ToggleScreenRequested(source="lirc"))
        await _settle(idle.service)
        await bus.emit_and_wait(UserActivity(source="web"))
        await _settle(idle.service)
        assert idle.ons() == 0


async def test_zero_minutes_means_never(bus: EventBus, tmp_path: Path) -> None:
    async with _Idle(bus, tmp_path, auto_off_minutes=0) as idle:
        await idle.wait(600)
        assert idle.offs() == 0


async def test_no_off_command_means_never(bus: EventBus, tmp_path: Path) -> None:
    async with _Idle(bus, tmp_path, off_command=None) as idle:
        await idle.wait(600)
        assert idle.offs() == 0


async def test_a_new_idle_time_applies_at_once(bus: EventBus, tmp_path: Path) -> None:
    async with _Idle(bus, tmp_path) as idle:
        idle.service.config.auto_off_minutes = 5  # what parent mode does
        await idle.wait(5)
        assert idle.offs() == 1


async def test_the_auto_off_is_silent(bus: EventBus, tmp_path: Path, fake_aplay: Path) -> None:
    async with _Idle(bus, tmp_path) as idle:
        idle.service.alsa_device = "softvol_effects"
        await idle.wait(15)
        assert idle.offs() == 1
    assert not Path(f"{fake_aplay}.wav").exists()