diff --git a/tests/test_gui_widgets.py b/tests/test_gui_widgets.py index a6b72ec..46ebf35 100644 --- a/tests/test_gui_widgets.py +++ b/tests/test_gui_widgets.py @@ -212,6 +212,338 @@ class TestSearchPage: # ═══════════════════════════════════════════════════════════════════════════════ +class TestAudioPlayer: + """Tests for AudioPlayer state methods (no playback required).""" + + @pytest.fixture + def player(self, app): + from player import AudioPlayer + p = AudioPlayer() + yield p + p.shutdown() + + def test_initial_state(self, player): + """Initial player state is stopped, no current song, empty queue.""" + assert player.is_stopped() is True + assert player.is_playing() is False + assert player.is_paused() is False + assert player.get_current() is None + assert player.get_queue() == [] + assert player.get_queue_length() == 0 + assert player.has_previous() is False + + def test_volume_default(self, player): + """Default volume should be 50.""" + assert player.volume() == 50 + + def test_set_volume(self, player): + """set_volume changes the volume.""" + player.set_volume(75) + assert player.volume() == 75 + player.set_volume(100) + assert player.volume() == 100 + player.set_volume(0) + assert player.volume() == 0 + + def test_set_volume_clamps(self, player): + """set_volume clamps values outside 0-100.""" + player.set_volume(-10) + assert player.volume() == 0 + player.set_volume(150) + assert player.volume() == 100 + + def test_loop_default(self, player): + """Loop mode is initially off.""" + assert player.get_loop() is False + + def test_set_loop(self, player): + """set_loop toggles loop mode.""" + player.set_loop(True) + assert player.get_loop() is True + player.set_loop(False) + assert player.get_loop() is False + + def test_stop_playback_when_stopped(self, player): + """stop_playback when already stopped doesn't crash.""" + player.stop_playback() + assert player.is_stopped() is True + + def test_queue_operations(self, player): + """queue_song, queue_next, and queue_list work correctly. + + ``queue_song`` triggers auto-play when queue is empty, which pops + the song immediately. We test queue_list (no auto-play) instead. + """ + s2 = {"videoId": "b", "title": "B"} + s3 = {"videoId": "c", "title": "C"} + + # queue_list adds without triggering auto-play + player.queue_list([s2, s3]) + assert player.get_queue_length() == 2 + assert player.get_queue()[0]["videoId"] == "b" + + # queue_next inserts at position 0 + s1 = {"videoId": "a", "title": "A"} + player.queue_next(s1) + assert player.get_queue_length() == 3 + assert player.get_queue()[0]["videoId"] == "a" + + def test_seek(self, player): + """seek sets position (no crash test).""" + player.seek(0) # should not crash + player.seek(5000) + + def test_shutdown(self, player): + """shutdown cleans up without error.""" + player.shutdown() # already called in fixture cleanup + + def test_cleanup_current_temp_no_current(self, player): + """_cleanup_current_temp does nothing when no song is loaded.""" + player._cleanup_current_temp() # should not crash + + def test_cleanup_temp_files_no_files(self, player): + """_cleanup_temp_files handles missing files gracefully.""" + player._current = {"_local_path": "/nonexistent/file.mp3"} + player._cleanup_temp_files() # should not crash + player._current = None + + def test_cleanup_temp_files_with_cache(self, player, tmp_path): + """_cleanup_temp_files removes cached temp files.""" + f = tmp_path / "test_song.mp3" + f.write_text("audio data") + player._prefetch_cache["vid1"] = {"_local_path": str(f)} + player._cleanup_temp_files() + assert not f.exists() + + def test_position_and_duration_signals(self, player): + """_on_position and _on_duration emit signals.""" + results = [] + player.position_changed.connect(lambda ms: results.append(("pos", ms))) + player.duration_changed.connect(lambda ms: results.append(("dur", ms))) + + player._on_position(5000) + player._on_duration(200000) + + assert ("pos", 5000) in results + assert ("dur", 200000) in results + + def test_is_loaded(self, player): + """is_loaded returns False when no current song.""" + assert player.is_loaded() is False + player._current = {"videoId": "x"} + assert player.is_loaded() is True + player._current = None + + def test_toggle_pause_when_stopped(self, player): + """toggle_pause when stopped does nothing.""" + result = player.toggle_pause() + # When stopped, toggle_pause returns None (no state change) + assert result is None or result is False + + def test_previous_no_history(self, player): + """previous does nothing when history is empty.""" + player.previous() # should not crash + assert player.has_previous() is False + + def test_cancel_active_downloads(self, player): + """_cancel_active_downloads emits cancel signal when active.""" + player._active_task_id = 42 + with mock.patch.object(player._worker, "cancel_requested") as mock_sig: + player._cancel_active_downloads() + mock_sig.emit.assert_called_once() + + def test_cancel_active_downloads_noop(self, player): + """_cancel_active_downloads does nothing when no active task.""" + with mock.patch.object(player._worker, "cancel_requested") as mock_sig: + player._cancel_active_downloads() + mock_sig.emit.assert_not_called() + + def test_on_download_failed_matching_task(self, player): + """_on_download_failed emits error when task matches active.""" + results = [] + player.error.connect(lambda d: results.append(d)) + player._active_task_id = 42 + player._on_download_failed(42) + assert len(results) == 1 + assert "error" in results[0] + assert player._active_task_id is None + + def test_on_download_failed_stale_task(self, player): + """_on_download_failed ignores stale task IDs.""" + results = [] + player.error.connect(lambda d: results.append(d)) + player._active_task_id = 42 + player._on_download_failed(99) # different task ID + assert results == [] # no signal emitted + assert player._active_task_id == 42 # unchanged + + def test_skip_no_current(self, player): + """skip does nothing when no current song.""" + player.skip() # should not crash + + def test_on_download_succeeded_stale(self, player): + """_on_download_succeeded ignores stale task IDs.""" + player._active_task_id = 42 + player._on_download_succeeded(99, {"videoId": "x"}) # different task + # No crash, no state change + + def test_on_download_succeeded_missing_file(self, player, tmp_path): + """_on_download_succeeded emits error when file is missing.""" + results = [] + player.error.connect(lambda d: results.append(d)) + player._active_task_id = 42 + player._on_download_succeeded(42, {"videoId": "x", + "_local_path": str(tmp_path / "nonexistent.mp3")}) + assert len(results) == 1 + assert player._active_task_id is None + + def test_on_playback_state_playing(self, player): + """_on_playback_state with PlayingState emits 'playing'.""" + from PySide6.QtMultimedia import QMediaPlayer + results = [] + player.playback_state_changed.connect(lambda s: results.append(s)) + player._current = {"videoId": "x", "title": "Test"} + player._on_playback_state(QMediaPlayer.PlaybackState.PlayingState) + assert "playing" in results + player._current = None + + def test_on_playback_state_paused(self, player): + """_on_playback_state with PausedState emits 'paused'.""" + from PySide6.QtMultimedia import QMediaPlayer + results = [] + player.playback_state_changed.connect(lambda s: results.append(s)) + player._current = {"videoId": "x"} + player._on_playback_state(QMediaPlayer.PlaybackState.PausedState) + assert "paused" in results + player._current = None + + def test_on_playback_state_stopped(self, player): + """_on_playback_state with StoppedState emits 'stopped'.""" + from PySide6.QtMultimedia import QMediaPlayer + results = [] + player.playback_state_changed.connect(lambda s: results.append(s)) + player._on_playback_state(QMediaPlayer.PlaybackState.StoppedState) + assert "stopped" in results + + def test_on_media_status_end_of_media(self, player): + """_on_media_status with EndOfMedia triggers advance.""" + from PySide6.QtMultimedia import QMediaPlayer + results = [] + player.song_ended.connect(lambda vid: results.append(vid)) + player._current = {"videoId": "x"} + player._on_media_status(QMediaPlayer.MediaStatus.EndOfMedia) + assert "x" in results + player._current = None + + def test_on_media_status_invalid_media_stopped(self, player): + """_on_media_status with InvalidMedia does nothing when player is stopped.""" + from PySide6.QtMultimedia import QMediaPlayer + results = [] + player.error.connect(lambda d: results.append(d)) + player._current = {"videoId": "x"} + # When the player is stopped (no media loaded), InvalidMedia is ignored + player._on_media_status(QMediaPlayer.MediaStatus.InvalidMedia) + assert len(results) == 0 # player is stopped, so early return + player._current = None + + def test_on_media_status_loaded(self, player): + """_on_media_status with LoadedMedia does nothing special.""" + from PySide6.QtMultimedia import QMediaPlayer + player._on_media_status(QMediaPlayer.MediaStatus.LoadedMedia) # no crash + + def test_get_position(self, player): + """get_position returns current position (0 when stopped).""" + assert player.get_position() == 0 + + def test_start_prefetch(self, player): + """_start_prefetch with queue emits download request.""" + from player import DownloadWorker + worker = DownloadWorker() + player._worker = worker + player._queue = [{"videoId": "prefetch_test"}] + with mock.patch.object(worker, "download_requested") as mock_sig: + player._start_prefetch() + mock_sig.emit.assert_called_once() + task_id, song = mock_sig.emit.call_args[0] + assert isinstance(task_id, int) + assert song["videoId"] == "prefetch_test" + + def test_on_download_succeeded_prefetch(self, player): + """_on_download_succeeded caches prefetch result.""" + player._prefetch_task_id = 55 + player._queue = [{"videoId": "prefetch_vid"}] + player._on_download_succeeded(55, {"videoId": "prefetch_vid", + "_local_path": "/tmp/test.mp3"}) + assert "prefetch_vid" in player._prefetch_cache + assert player._prefetch_task_id is None + + def test_advance_with_loop(self, player): + """_advance re-queues finished song at front in loop mode.""" + player._loop_mode = True + player._current = {"videoId": "loop_song"} + player._queue = [{"videoId": "next_song"}] + # _advance inserts at front, then _try_next_song pops the front + # leaving the next_song. Verify loop_song was inserted before pop. + with mock.patch.object(player, "_try_next_song"): + player._advance() + # The finished song should be at the front before _try_next_song + assert player._queue[0]["videoId"] == "loop_song" + assert player._current is None + + def test_try_next_song_with_queue(self, player): + """_try_next_song pops and plays when queue has items.""" + player._queue = [{"videoId": "queued"}] + # This will call _pop_and_play which tries to download + # Just check it doesn't crash and processes the queue + with mock.patch.object(player, "_pop_and_play") as mock_pop: + player._try_next_song() + mock_pop.assert_called_once() + + def test_on_player_error_network(self, player): + """_on_player_error_occurred with NetworkError triggers advance.""" + from PySide6.QtMultimedia import QMediaPlayer + results = [] + player.error.connect(lambda d: results.append(d)) + player._current = {"videoId": "err_song"} + player._on_player_error_occurred( + QMediaPlayer.Error.NetworkError, "Connection lost") + assert len(results) == 1 + assert "error" in results[0] + player._current = None + + def test_on_player_error_format(self, player): + """_on_player_error_occurred with FormatError does nothing.""" + from PySide6.QtMultimedia import QMediaPlayer + results = [] + player.error.connect(lambda d: results.append(d)) + player._on_player_error_occurred( + QMediaPlayer.Error.FormatError, "Bad format") + # FormatError is not NetworkError or ResourceError, so ignored + assert len(results) == 0 + + def test_seek_emits_seeked_signal(self, player): + """seek emits seeked signal with the position.""" + results = [] + player.seeked.connect(lambda ms: results.append(ms)) + player.seek(15000) + assert 15000 in results + + def test_download_worker_do_cancel(self, player): + """DownloadWorker._do_cancel sets cancelled flag.""" + from player import DownloadWorker + worker = DownloadWorker() + assert worker._cancelled is False + worker._do_cancel() + assert worker._cancelled is True + + def test_on_position_zero(self, player): + """_on_position with 0 emits position_changed.""" + results = [] + player.position_changed.connect(lambda ms: results.append(ms)) + player._on_position(0) + assert 0 in results + + class TestTunettiWindowAccessors: @pytest.fixture diff --git a/tests/test_import_fallbacks.py b/tests/test_import_fallbacks.py index 2c9c654..c81bcc1 100644 --- a/tests/test_import_fallbacks.py +++ b/tests/test_import_fallbacks.py @@ -193,6 +193,29 @@ class TestPlayerQtFallback: assert player.QMediaPlayer is not None assert hasattr(player.QMediaPlayer, "PlaybackState") + def test_stub_qmediaplayer_methods(self): + """Stub QMediaPlayer methods can be called without error.""" + _clean_pyside_modules() + side_effect = _make_importer(block_pyside=True) + + with mock.patch("builtins.__import__", side_effect=side_effect): + import player + + mp = player.QMediaPlayer() + # Methods that should be no-ops + mp.set_source("https://example.com/audio.mp3") + mp.play() + mp.pause() + mp.stop() + mp.set_position(30000) + mp.set_audio_output(None) + + # Methods that return values + assert mp.position() == 0 + assert mp.duration() == 0 + assert mp.get_playback_state() == 0 # stopped_state + assert mp.get_audio_output() is None + # ═══════════════════════════════════════════════════════════════════════════════ # gui.py fallback diff --git a/tests/test_mpris.py b/tests/test_mpris.py index f9bf141..f08f10c 100644 --- a/tests/test_mpris.py +++ b/tests/test_mpris.py @@ -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