feat(media): MprisController.pause() via dbus-next

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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

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:
-    async def pause(self) -> None:
+    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:
+        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