diff --git a/tests/test_media_mpris.py b/tests/test_media_mpris.py new file mode 100644 index 0000000..e8dc1bd --- /dev/null +++ b/tests/test_media_mpris.py @@ -0,0 +1,63 @@ +"""Tests für MprisController (Linux/MPRIS).""" + +import sys +from unittest.mock import AsyncMock, MagicMock + +import pytest + +pytestmark = pytest.mark.skipif(sys.platform != "linux", reason="MPRIS is Linux-only") + + +def _make_player(status: str) -> MagicMock: + """Erzeugt einen gemockten Player-Proxy mit gegebenem PlaybackStatus.""" + player = MagicMock() + player.get_playback_status = AsyncMock(return_value=status) + player.call_pause = AsyncMock() + player.call_play = AsyncMock() + return player + + +@pytest.mark.asyncio +async def test_pause_with_no_players_is_noop(monkeypatch): + from whisper_local.media._mpris import MprisController + + controller = MprisController() + monkeypatch.setattr( + controller, "_list_player_names", AsyncMock(return_value=[]) + ) + monkeypatch.setattr( + controller, "_get_player_interface", AsyncMock() + ) + + await controller.pause() + + assert controller._paused == [] + + +@pytest.mark.asyncio +async def test_pause_pauses_only_playing_players(monkeypatch): + from whisper_local.media._mpris import MprisController + + playing = _make_player("Playing") + paused = _make_player("Paused") + + async def fake_get_player(name: str): + return {"org.mpris.MediaPlayer2.spotify": playing, + "org.mpris.MediaPlayer2.vlc": paused}[name] + + controller = MprisController() + monkeypatch.setattr( + controller, + "_list_player_names", + AsyncMock(return_value=[ + "org.mpris.MediaPlayer2.spotify", + "org.mpris.MediaPlayer2.vlc", + ]), + ) + monkeypatch.setattr(controller, "_get_player_interface", fake_get_player) + + await controller.pause() + + playing.call_pause.assert_awaited_once() + paused.call_pause.assert_not_awaited() + assert controller._paused == ["org.mpris.MediaPlayer2.spotify"] diff --git a/whisper_local/media/_mpris.py b/whisper_local/media/_mpris.py index 201160e..8f102b7 100644 --- a/whisper_local/media/_mpris.py +++ b/whisper_local/media/_mpris.py @@ -1,13 +1,65 @@ -"""Linux-MPRIS-Implementierung. Vollständige Logik folgt in späteren Tasks.""" +"""Linux-MPRIS-Implementierung via dbus-next.""" +import asyncio import logging +from typing import Any logger = logging.getLogger(__name__) +MPRIS_PREFIX = "org.mpris.MediaPlayer2." +MPRIS_PATH = "/org/mpris/MediaPlayer2" +PLAYER_IFACE = "org.mpris.MediaPlayer2.Player" +DBUS_SERVICE = "org.freedesktop.DBus" +DBUS_PATH = "/org/freedesktop/DBus" +DBUS_IFACE = "org.freedesktop.DBus" + class MprisController: + def __init__(self) -> None: + self._paused: list[str] = [] + self._bus: Any = None + + async def _ensure_bus(self) -> Any: + if self._bus is None: + from dbus_next.aio import MessageBus + self._bus = await MessageBus().connect() + return self._bus + + async def _list_player_names(self) -> list[str]: + bus = await self._ensure_bus() + intro = await bus.introspect(DBUS_SERVICE, DBUS_PATH) + obj = bus.get_proxy_object(DBUS_SERVICE, DBUS_PATH, intro) + iface = obj.get_interface(DBUS_IFACE) + names = await iface.call_list_names() + return [n for n in names if n.startswith(MPRIS_PREFIX)] + + async def _get_player_interface(self, bus_name: str) -> Any: + bus = await self._ensure_bus() + intro = await bus.introspect(bus_name, MPRIS_PATH) + obj = bus.get_proxy_object(bus_name, MPRIS_PATH, intro) + return obj.get_interface(PLAYER_IFACE) + + async def _pause_player(self, bus_name: str) -> str | None: + """Pausiert einen Player, wenn er im Status 'Playing' ist. + Gibt den Bus-Namen zurück, wenn pausiert wurde, sonst None. + """ + player = await self._get_player_interface(bus_name) + status = await player.get_playback_status() + if status != "Playing": + return None + await player.call_pause() + return bus_name + async def pause(self) -> None: - return None + names = await self._list_player_names() + results = await asyncio.gather( + *(self._pause_player(n) for n in names), + return_exceptions=True, + ) + self._paused = [ + name for name in results + if isinstance(name, str) + ] async def resume(self) -> None: return None