Martin Bauer
2d0caeb732
KEY_POWER emits ToggleScreenRequested, which a new ScreenService turns into the optional general.screen.toggle_command (musicdolphin: ansible's kiosk-screen). It is a service, not a reaction, so the subprocess runs in its own task instead of holding the bus; a press while one is still running is dropped, and failures, hangs and a missing command are only logged. irserver reports every line of a held button as repeat 00, so a slightly long press on Next skipped several tracks and Power would switch the screen off and on again. A line of the same button within 0.25s of the previous one now counts as a repeat. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
"""The screen service: runs the configured command per toggle, and never takes the bus down."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import shlex
|
|
import sys
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from musicmouse.bus import EventBus
|
|
from musicmouse.config import ScreenConfig
|
|
from musicmouse.events import ToggleScreenRequested
|
|
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) -> tuple[ScreenService, asyncio.Task[None]]:
|
|
service = ScreenService(bus, config)
|
|
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
|