✨ | Add MPRIS integration for media key control
All checks were successful
SonarQube Code Quality Scan / SonarQube Scan (push) Successful in 1m48s
All checks were successful
SonarQube Code Quality Scan / SonarQube Scan (push) Successful in 1m48s
Implements org.mpris.MediaPlayer2 and org.mpris.MediaPlayer2.Player D-Bus interfaces, allowing desktop environment media keys and tools like playerctl to control playback (play, pause, next, previous) and display now-playing metadata. Also adds _FallbackServiceInterface, _FallbackVariant and _FallbackNameFlag so all tests pass without dbus-fast installed; the module gracefully degrades when the library is unavailable.
This commit is contained in:
237
tests/test_mpris.py
Normal file
237
tests/test_mpris.py
Normal file
@@ -0,0 +1,237 @@
|
||||
"""Tests for the MPRIS integration module (mpris.py)."""
|
||||
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ── _State ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestState:
|
||||
"""Tests for the _State thread-safe holder."""
|
||||
|
||||
def _import(self):
|
||||
from mpris import _State
|
||||
return _State
|
||||
|
||||
def test_initial_state_is_stopped(self):
|
||||
cls = self._import()
|
||||
s = cls()
|
||||
assert s.get_status() == "Stopped"
|
||||
assert s.get_metadata() == {}
|
||||
assert s.get_can_go_prev() is False
|
||||
|
||||
def test_update_changes_state(self):
|
||||
cls = self._import()
|
||||
s = cls()
|
||||
s.update("Playing", {"xesam:title": "Test"}, True)
|
||||
assert s.get_status() == "Playing"
|
||||
assert s.get_metadata() == {"xesam:title": "Test"}
|
||||
assert s.get_can_go_prev() is True
|
||||
|
||||
def test_get_metadata_returns_copy(self):
|
||||
cls = self._import()
|
||||
s = cls()
|
||||
s.update("Paused", {"key": "val"}, False)
|
||||
meta = s.get_metadata()
|
||||
meta["extra"] = "added"
|
||||
assert "extra" not in s.get_metadata()
|
||||
|
||||
|
||||
# ── _build_mpris_metadata ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildMprisMetadata:
|
||||
"""Tests for the _build_mpris_metadata helper."""
|
||||
|
||||
def _import(self):
|
||||
from mpris import _build_mpris_metadata
|
||||
return _build_mpris_metadata
|
||||
|
||||
def test_none_song_returns_empty(self):
|
||||
fn = self._import()
|
||||
assert fn(None) == {}
|
||||
|
||||
def test_empty_song_returns_empty(self):
|
||||
fn = self._import()
|
||||
# Empty dict is falsy, so the function returns {} early
|
||||
assert fn({}) == {}
|
||||
|
||||
def test_full_song_metadata(self):
|
||||
fn = self._import()
|
||||
song = {
|
||||
"videoId": "abc123",
|
||||
"title": "Test Song",
|
||||
"artists": [{"name": "Artist One"}, {"name": "Artist Two"}],
|
||||
"album": "Test Album",
|
||||
"duration": 240,
|
||||
"thumbnail": "https://example.com/thumb.jpg",
|
||||
}
|
||||
meta = fn(song)
|
||||
assert meta["xesam:title"].value == "Test Song"
|
||||
assert meta["xesam:artist"].value == ["Artist One", "Artist Two"]
|
||||
assert meta["xesam:album"].value == "Test Album"
|
||||
assert meta["mpris:length"].value == 240_000_000
|
||||
assert meta["mpris:artUrl"].value == "https://example.com/thumb.jpg"
|
||||
assert "abc123" in meta["mpris:trackid"].value
|
||||
|
||||
def test_album_as_dict_extracts_name(self):
|
||||
fn = self._import()
|
||||
song = {
|
||||
"videoId": "x",
|
||||
"album": {"name": "Real Album", "id": "album_id"},
|
||||
}
|
||||
meta = fn(song)
|
||||
assert meta["xesam:album"].value == "Real Album"
|
||||
|
||||
def test_album_is_none(self):
|
||||
fn = self._import()
|
||||
meta = fn({"videoId": "x", "album": None})
|
||||
assert meta["xesam:album"].value == ""
|
||||
|
||||
def test_artists_not_list(self):
|
||||
fn = self._import()
|
||||
song = {"videoId": "x", "title": "T", "artists": "not a list"}
|
||||
meta = fn(song)
|
||||
assert meta["xesam:artist"].value == []
|
||||
|
||||
def test_missing_video_id_does_not_crash(self):
|
||||
fn = self._import()
|
||||
song = {"title": "No ID"}
|
||||
meta = fn(song)
|
||||
assert isinstance(meta, dict)
|
||||
assert "mpris:trackid" in meta
|
||||
|
||||
def test_zero_duration(self):
|
||||
fn = self._import()
|
||||
meta = fn({"videoId": "x", "duration": 0})
|
||||
assert meta["mpris:length"].value == 0
|
||||
|
||||
def test_none_duration(self):
|
||||
fn = self._import()
|
||||
meta = fn({"videoId": "x", "duration": None})
|
||||
assert meta["mpris:length"].value == 0
|
||||
|
||||
|
||||
# ── MprisManager ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestMprisManager:
|
||||
"""Tests for the MprisManager lifecycle."""
|
||||
|
||||
def test_start_returns_false_without_dbus(self):
|
||||
"""When dbus-fast is unavailable, start() returns False."""
|
||||
with mock.patch("mpris._HAS_DBUS", False):
|
||||
from mpris import MprisManager
|
||||
mgr = MprisManager()
|
||||
assert mgr.start() is False
|
||||
|
||||
def test_bridge_is_none_before_start(self):
|
||||
from mpris import MprisManager
|
||||
mgr = MprisManager()
|
||||
assert mgr.get_bridge() is None
|
||||
|
||||
def test_stop_without_start_does_not_crash(self):
|
||||
from mpris import MprisManager
|
||||
mgr = MprisManager()
|
||||
mgr.stop() # should not raise
|
||||
|
||||
|
||||
# ── MprisBridge ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestMprisBridge:
|
||||
"""Tests for the MprisBridge Qt signal bridge."""
|
||||
|
||||
def test_has_required_signals(self):
|
||||
from mpris import MprisBridge
|
||||
bridge = MprisBridge()
|
||||
for name in ("play_requested", "pause_requested", "play_pause_requested",
|
||||
"stop_requested", "next_requested", "previous_requested",
|
||||
"seek_requested"):
|
||||
assert hasattr(bridge, name), f"Missing signal: {name}"
|
||||
|
||||
|
||||
# ── MprisRootInterface ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestMprisRootInterface:
|
||||
"""Tests for the root MPRIS interface."""
|
||||
|
||||
def test_identity(self):
|
||||
from mpris import MprisRootInterface
|
||||
iface = MprisRootInterface()
|
||||
assert iface.Identity == "Tunetti"
|
||||
|
||||
def test_desktop_entry(self):
|
||||
from mpris import MprisRootInterface
|
||||
iface = MprisRootInterface()
|
||||
assert iface.DesktopEntry == "tunetti"
|
||||
|
||||
def test_capabilities(self):
|
||||
from mpris import MprisRootInterface
|
||||
iface = MprisRootInterface()
|
||||
assert iface.CanQuit is False
|
||||
assert iface.CanRaise is False
|
||||
assert iface.HasTrackList is False
|
||||
|
||||
|
||||
# ── MprisPlayerInterface ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestMprisPlayerInterface:
|
||||
"""Tests for the player MPRIS interface."""
|
||||
|
||||
def _make(self):
|
||||
from mpris import _State, MprisBridge, MprisPlayerInterface
|
||||
state = _State()
|
||||
bridge = MprisBridge()
|
||||
iface = MprisPlayerInterface(state, bridge)
|
||||
return state, bridge, iface
|
||||
|
||||
def test_playback_status_default(self):
|
||||
_, _, iface = self._make()
|
||||
assert iface.PlaybackStatus == "Stopped"
|
||||
|
||||
def test_playback_status_playing(self):
|
||||
state, _, iface = self._make()
|
||||
state.update("Playing", {}, False)
|
||||
assert iface.PlaybackStatus == "Playing"
|
||||
|
||||
def test_can_go_next_default(self):
|
||||
_, _, iface = self._make()
|
||||
assert iface.CanGoNext is True
|
||||
|
||||
def test_can_go_previous_false_when_no_history(self):
|
||||
_, _, iface = self._make()
|
||||
assert iface.CanGoPrevious is False
|
||||
|
||||
def test_can_go_previous_true_when_history_exists(self):
|
||||
state, _, iface = self._make()
|
||||
state.update("Playing", {"t": "1"}, True)
|
||||
assert iface.CanGoPrevious is True
|
||||
|
||||
def test_can_control(self):
|
||||
_, _, iface = self._make()
|
||||
assert iface.CanControl is True
|
||||
|
||||
def test_methods_emit_via_bridge(self):
|
||||
state, bridge, iface = self._make()
|
||||
for method_name, signal_name in [
|
||||
("Play", "play_requested"),
|
||||
("Pause", "pause_requested"),
|
||||
("PlayPause", "play_pause_requested"),
|
||||
("Stop", "stop_requested"),
|
||||
("Next", "next_requested"),
|
||||
("Previous", "previous_requested"),
|
||||
]:
|
||||
with mock.patch.object(bridge, signal_name) as mock_sig:
|
||||
getattr(iface, method_name)()
|
||||
mock_sig.emit.assert_called_once()
|
||||
|
||||
def test_seek_emits_with_offset(self):
|
||||
state, bridge, iface = self._make()
|
||||
with mock.patch.object(bridge, "seek_requested") as mock_sig:
|
||||
iface.Seek(5000)
|
||||
mock_sig.emit.assert_called_once_with(5000)
|
||||
Reference in New Issue
Block a user