Tunetti/tests/test_mpris.py

"""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):
        _, 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):
        _, 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)


# ── 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