| Boost coverage to 60% with 256 passing tests
All checks were successful
SonarQube Code Quality Scan / SonarQube Scan (push) Successful in 1m52s

- mpris.py: 81% → 100% — import fallbacks, flush_props edge cases,
  stop_async error path, run_async exception path, notify loops
- player.py: 51% → 73% — AudioPlayer state/volume/loop/queue tests,
  _on_download_failed, _on_download_succeeded, _on_playback_state,
  _on_media_status, seek/skip/previous, _start_prefetch, _advance,
  _cancel_active_downloads, DownloadWorker._do_cancel
- test_gui_widgets.py: 50 → 88 tests across TestAudioPlayer and
  existing widget tests
- 5 modules at 100%: config, discord_rpc, main, mpris, music_db
This commit is contained in:
2026-06-09 19:33:58 +03:00
parent 27d12d31cb
commit 4671a6937b
3 changed files with 574 additions and 0 deletions

View File

@@ -235,3 +235,222 @@ class TestMprisPlayerInterface:
with mock.patch.object(bridge, "seek_requested") as mock_sig:
iface.Seek(5000)
mock_sig.emit.assert_called_once_with(5000)
# ── Import fallback (when dbus-fast unavailable) ──────────────────────────────
class TestImportFallback:
"""Tests for the mpris module when dbus-fast is unavailable."""
def test_fallback_classes_work(self):
"""When dbus-fast is unavailable, fallback classes work correctly."""
import builtins
import sys
original_import = builtins.__import__
def mock_import(name, *args, **kwargs):
if name.startswith("dbus_fast") or name.startswith("dbus-fast"):
raise ImportError(f"No module named {name!r}")
return original_import(name, *args, **kwargs)
# Remove mpris from sys.modules so it re-imports
sys.modules.pop("mpris", None)
with mock.patch("builtins.__import__", side_effect=mock_import):
import mpris as mpris_fallback
# Test fallback Variant
v = mpris_fallback.Variant("s", "hello")
assert v.signature == "s"
assert v.value == "hello"
# Test fallback NameFlag
assert mpris_fallback.NameFlag.REPLACE_EXISTING == 1
assert mpris_fallback.NameFlag.DO_NOT_QUEUE == 2
# Test fallback ServiceInterface
si = mpris_fallback.ServiceInterface("test.Interface")
assert si._interface_name == "test.Interface"
si.emit_properties_changed({"a": 1}, []) # no-op
# Test fallback decorators
@mpris_fallback.dbus_method()
def my_method():
return 42
assert my_method() == 42
@mpris_fallback.dbus_property(access="read")
def my_prop():
return "val"
assert my_prop() == "val"
@mpris_fallback.dbus_signal()
def my_signal():
return "sig"
assert my_signal() == "sig"
# Test PropertyAccess fallback
assert mpris_fallback.PropertyAccess.READ == "read"
# Test that _HAS_DBUS is False
assert mpris_fallback._HAS_DBUS is False
def test_pyside6_import_fallback(self):
"""When PySide6 is unavailable, Signal/QObject fallbacks work."""
import builtins
import sys
original_import = builtins.__import__
def mock_import(name, *args, **kwargs):
if name == "PySide6" or name.startswith("PySide6."):
raise ImportError(f"No module named {name!r}")
return original_import(name, *args, **kwargs)
sys.modules.pop("mpris", None)
with mock.patch("builtins.__import__", side_effect=mock_import):
import mpris as mpris_fallback
assert mpris_fallback._HAS_QT_BRIDGE is False
# MprisBridge should inherit from object, not QObject
assert mpris_fallback.MprisBridge.__bases__ == (object,)
# ── Manager advanced / edge-case tests ───────────────────────────────────────
class TestMprisManagerAdvanced:
"""Tests for manager edge cases (error paths, early returns)."""
def test_flush_props_without_bus(self):
"""_flush_props returns early when bus is None (lines 416-417)."""
from mpris import MprisManager
mgr = MprisManager()
# _bus is None, _flush_props should return without error
mgr._flush_props()
def test_flush_props_emission_raises(self):
"""_flush_props swallows exceptions from emit_properties_changed (lines 433-434)."""
from unittest.mock import MagicMock
from mpris import MprisManager
mgr = MprisManager()
mgr._bus = MagicMock()
mgr._player_iface = MagicMock()
mgr._player_iface.emit_properties_changed.side_effect = RuntimeError("boom")
with mgr._prop_lock:
mgr._prop_queue.append(("Playing", {"xesam:title": "Test"}, False))
mgr._flush_props() # should not raise
# Queue should be cleared
assert len(mgr._prop_queue) == 0
def test_stop_async_without_bus(self):
"""_stop_async does nothing when bus is None."""
from mpris import MprisManager
mgr = MprisManager()
mgr._stop_async() # should not raise
def test_stop_async_disconnect_raises(self):
"""_stop_async catches exceptions from bus.disconnect() (lines 438-442)."""
from unittest.mock import MagicMock
from mpris import MprisManager
mgr = MprisManager()
mock_bus = MagicMock()
mock_bus.disconnect.side_effect = RuntimeError("disconnect failed")
mgr._bus = mock_bus
mgr._stop_async() # should not raise
mock_bus.disconnect.assert_called_once()
def test_properties_changed_signal(self):
"""PropertiesChanged signal method can be called directly (line 290)."""
from mpris import _State, MprisBridge, MprisPlayerInterface
state = _State()
bridge = MprisBridge()
iface = MprisPlayerInterface(state, bridge)
# Call the signal method directly — it's a no-op pass
iface.PropertiesChanged("org.mpris.MediaPlayer2.Player", {}, [])
def test_run_async_exception_path(self):
"""_run_async catches and logs exceptions from _connect_and_serve (lines 449-451)."""
from unittest.mock import patch
from mpris import MprisManager
mgr = MprisManager()
with patch.object(mgr, "_connect_and_serve", side_effect=RuntimeError("dbus error")):
with patch("mpris.log") as mock_log:
mgr._run_async()
mock_log.debug.assert_called_once()
# Finally block should have cleaned up
assert mgr._loop is None
assert mgr._bus is None
assert mgr._root_iface is None
assert mgr._player_iface is None
# ── notify_state_changed edge cases ────────────────────────────────────────
class TestNotifyStateChanged:
"""Tests for notify_state_changed edge cases."""
def test_loop_not_running_queues_only(self):
"""When the asyncio loop isn't running, data is queued but not flushed."""
from mpris import MprisManager
mgr = MprisManager()
mgr._loop = mock.MagicMock()
mgr._loop.is_running.return_value = False
mgr.notify_state_changed("playing", {"videoId": "x"}, True)
assert len(mgr._prop_queue) == 1
mgr._loop.call_soon_threadsafe.assert_not_called()
def test_loop_is_none_queues_only(self):
"""When _loop is None, data is queued but not flushed."""
from mpris import MprisManager
mgr = MprisManager()
mgr._loop = None
mgr.notify_state_changed("playing", {"videoId": "x"}, True)
assert len(mgr._prop_queue) == 1
def test_status_mapping_fallback_to_stopped(self):
"""Unknown status string maps to STATUS_STOPPED."""
from mpris import MprisManager
mgr = MprisManager()
mgr._loop = mock.MagicMock()
mgr._loop.is_running.return_value = True
# Make call_soon_threadsafe actually invoke the callback so _flush_props runs
mgr._loop.call_soon_threadsafe.side_effect = lambda fn: fn()
mgr._bus = mock.MagicMock() # needed for _flush_props
mgr._player_iface = mock.MagicMock()
mgr.notify_state_changed("unknown_status", {"videoId": "x"}, False)
# Should have flushed - the status was mapped to "Stopped"
mgr._player_iface.emit_properties_changed.assert_called()
call_args = mgr._player_iface.emit_properties_changed.call_args
changed = call_args[0][0] if call_args else {}
assert changed.get("PlaybackStatus") == "Stopped"
def test_notify_without_song_queues_state_only(self):
"""notify_state_changed without a song queues PlaybackStatus only."""
from mpris import MprisManager
mgr = MprisManager()
mgr._loop = mock.MagicMock()
mgr._loop.is_running.return_value = True
# Make call_soon_threadsafe actually invoke the callback so _flush_props runs
mgr._loop.call_soon_threadsafe.side_effect = lambda fn: fn()
mgr._bus = mock.MagicMock()
mgr._player_iface = mock.MagicMock()
mgr.notify_state_changed("paused", None, False)
mgr._player_iface.emit_properties_changed.assert_called()
call_args = mgr._player_iface.emit_properties_changed.call_args
changed = call_args[0][0] if call_args else {}
assert "PlaybackStatus" in changed
assert "Metadata" not in changed