159 lines
4.9 KiB
Python
159 lines
4.9 KiB
Python
"""Regression tests: MPRIS property reads must be issued as raw
|
|
org.freedesktop.DBus.Properties.Get messages, independent of introspection.
|
|
"""
|
|
|
|
import asyncio
|
|
from typing import Callable, Optional
|
|
|
|
import pytest
|
|
from dbus_next.constants import MessageType
|
|
from dbus_next.message import Message
|
|
from dbus_next.signature import Variant
|
|
|
|
from lrx_cli.models import TrackMeta
|
|
from lrx_cli.mpris import _fetch_metadata_dbus, _get_playback_status
|
|
from lrx_cli.watch.player import PlayerMonitor
|
|
|
|
PLAYER = "org.mpris.MediaPlayer2.generic1"
|
|
|
|
PLAYER_METADATA = {
|
|
"mpris:trackid": Variant("o", "/org/example/Tracks/42"),
|
|
"mpris:length": Variant("x", 239_761_000),
|
|
"mpris:artUrl": Variant("s", "http://example.com/cover.jpg"),
|
|
"xesam:album": Variant("s", "album"),
|
|
"xesam:artist": Variant("as", ["artist"]),
|
|
"xesam:title": Variant("s", "title"),
|
|
"xesam:url": Variant("s", "http://example.com/stream"),
|
|
}
|
|
|
|
|
|
class FakeBus:
|
|
"""Test double exposing only the low-level call/disconnect surface."""
|
|
|
|
def __init__(self, reply: Callable[[Message], Optional[Message]]) -> None:
|
|
self._reply = reply
|
|
self.calls: list[Message] = []
|
|
self.disconnected = False
|
|
|
|
async def call(self, message: Message) -> Optional[Message]:
|
|
self.calls.append(message)
|
|
return self._reply(message)
|
|
|
|
def disconnect(self) -> None:
|
|
self.disconnected = True
|
|
|
|
|
|
def _return(signature: str, body: list) -> Message:
|
|
return Message(
|
|
message_type=MessageType.METHOD_RETURN,
|
|
reply_serial=1,
|
|
signature=signature,
|
|
body=body,
|
|
)
|
|
|
|
|
|
def _names_reply(names: list[str]) -> Message:
|
|
return _return("as", [names])
|
|
|
|
|
|
def _metadata_reply() -> Message:
|
|
return _return("v", [Variant("a{sv}", PLAYER_METADATA)])
|
|
|
|
|
|
def _state_reply(msg: Message) -> Optional[Message]:
|
|
"""Reply to ListNames and the PlaybackStatus/Metadata property reads."""
|
|
if msg.member == "ListNames":
|
|
return _names_reply([PLAYER])
|
|
if msg.member == "Get":
|
|
if msg.body[1] == "PlaybackStatus":
|
|
return _return("v", [Variant("s", "Playing")])
|
|
if msg.body[1] == "Metadata":
|
|
return _metadata_reply()
|
|
return None
|
|
|
|
|
|
def _monitor(bus: FakeBus) -> PlayerMonitor:
|
|
monitor = PlayerMonitor(
|
|
on_players_changed=lambda: None,
|
|
on_seeked=lambda bus_name, position_ms: None,
|
|
on_playback_status=lambda bus_name, status: None,
|
|
player_blacklist=(),
|
|
)
|
|
monitor._bus = bus # type: ignore[assignment]
|
|
return monitor
|
|
|
|
|
|
def test_get_playback_status_sends_raw_properties_get() -> None:
|
|
def reply(msg: Message) -> Optional[Message]:
|
|
if msg.member == "Get":
|
|
return _return("v", [Variant("s", "Playing")])
|
|
return None
|
|
|
|
bus = FakeBus(reply)
|
|
status = asyncio.run(_get_playback_status(bus, PLAYER)) # type: ignore[arg-type]
|
|
|
|
assert status == "Playing"
|
|
request = bus.calls[0]
|
|
assert request.destination == PLAYER
|
|
assert request.path == "/org/mpris/MediaPlayer2"
|
|
assert request.interface == "org.freedesktop.DBus.Properties"
|
|
assert request.member == "Get"
|
|
assert request.body == ["org.mpris.MediaPlayer2.Player", "PlaybackStatus"]
|
|
|
|
|
|
def test_fetch_metadata_dbus_reads_metadata_via_raw_properties_get() -> None:
|
|
bus = FakeBus(_state_reply)
|
|
track = asyncio.run(_fetch_metadata_dbus(None, "", (), bus=bus)) # type: ignore[arg-type]
|
|
|
|
# trackid without a Spotify prefix normalizes to None
|
|
assert track == TrackMeta(
|
|
trackid=None,
|
|
length=239_761,
|
|
album="album",
|
|
artist="artist",
|
|
title="title",
|
|
url="http://example.com/stream",
|
|
)
|
|
assert not any(msg.member == "Introspect" for msg in bus.calls)
|
|
# injected bus stays owned by the caller
|
|
assert bus.disconnected is False
|
|
|
|
|
|
def test_fetch_metadata_dbus_disconnects_self_created_bus(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
bus = FakeBus(_state_reply)
|
|
|
|
class FakeMessageBus:
|
|
@staticmethod
|
|
async def connect() -> FakeBus:
|
|
return bus
|
|
|
|
monkeypatch.setattr("lrx_cli.mpris.MessageBus", lambda bus_type: FakeMessageBus)
|
|
track = asyncio.run(_fetch_metadata_dbus(None, "", ()))
|
|
|
|
assert track is not None
|
|
assert track.title == "title"
|
|
assert bus.disconnected is True
|
|
|
|
|
|
def test_player_monitor_refresh_reads_state_via_raw_properties_get() -> None:
|
|
monitor = _monitor(FakeBus(_state_reply))
|
|
asyncio.run(monitor.refresh())
|
|
|
|
state = monitor.players.get(PLAYER)
|
|
assert state is not None
|
|
assert state.status == "Playing"
|
|
assert state.track is not None
|
|
assert state.track.title == "title"
|
|
|
|
|
|
def test_player_monitor_get_position_ms_reads_raw_position() -> None:
|
|
def reply(msg: Message) -> Optional[Message]:
|
|
if msg.member == "Get":
|
|
return _return("v", [Variant("x", 231_008_000)])
|
|
return None
|
|
|
|
monitor = _monitor(FakeBus(reply))
|
|
assert asyncio.run(monitor.get_position_ms(PLAYER)) == 231_008
|