whisper-local/tests/test_media_mpris.py

"""Tests für MprisController (Linux/MPRIS)."""

import sys
from unittest.mock import AsyncMock, MagicMock

import pytest

pytestmark = pytest.mark.skipif(sys.platform != "linux", reason="MPRIS is Linux-only")


def _make_player(status: str) -> MagicMock:
    """Erzeugt einen gemockten Player-Proxy mit gegebenem PlaybackStatus."""
    player = MagicMock()
    player.get_playback_status = AsyncMock(return_value=status)
    player.call_pause = AsyncMock()
    player.call_play = AsyncMock()
    return player


@pytest.mark.asyncio
async def test_pause_with_no_players_is_noop(monkeypatch):
    from whisper_local.media._mpris import MprisController

    controller = MprisController()
    monkeypatch.setattr(
        controller, "_list_player_names", AsyncMock(return_value=[])
    )
    monkeypatch.setattr(
        controller, "_get_player_interface", AsyncMock()
    )

    await controller.pause()

    assert controller._paused == []


@pytest.mark.asyncio
async def test_pause_pauses_only_playing_players(monkeypatch):
    from whisper_local.media._mpris import MprisController

    playing = _make_player("Playing")
    paused = _make_player("Paused")

    async def fake_get_player(name: str):
        return {"org.mpris.MediaPlayer2.spotify": playing,
                "org.mpris.MediaPlayer2.vlc": paused}[name]

    controller = MprisController()
    monkeypatch.setattr(
        controller,
        "_list_player_names",
        AsyncMock(return_value=[
            "org.mpris.MediaPlayer2.spotify",
            "org.mpris.MediaPlayer2.vlc",
        ]),
    )
    monkeypatch.setattr(controller, "_get_player_interface", fake_get_player)

    await controller.pause()

    playing.call_pause.assert_awaited_once()
    paused.call_pause.assert_not_awaited()
    assert controller._paused == ["org.mpris.MediaPlayer2.spotify"]


@pytest.mark.asyncio
async def test_resume_plays_only_previously_paused(monkeypatch):
    from whisper_local.media._mpris import MprisController

    spotify = _make_player("Paused")
    vlc = _make_player("Paused")

    async def fake_get_player(name: str):
        return {"org.mpris.MediaPlayer2.spotify": spotify,
                "org.mpris.MediaPlayer2.vlc": vlc}[name]

    controller = MprisController()
    controller._paused = ["org.mpris.MediaPlayer2.spotify"]
    monkeypatch.setattr(controller, "_get_player_interface", fake_get_player)

    await controller.resume()

    spotify.call_play.assert_awaited_once()
    vlc.call_play.assert_not_awaited()
    assert controller._paused == []


@pytest.mark.asyncio
async def test_resume_with_empty_paused_list_is_noop(monkeypatch):
    from whisper_local.media._mpris import MprisController

    controller = MprisController()
    controller._paused = []
    get_player = AsyncMock()
    monkeypatch.setattr(controller, "_get_player_interface", get_player)

    await controller.resume()

    get_player.assert_not_awaited()


@pytest.mark.asyncio
async def test_pause_logs_and_continues_when_single_player_fails(monkeypatch, caplog):
    from whisper_local.media._mpris import MprisController

    good = _make_player("Playing")

    async def fake_get_player(name: str):
        if name.endswith("broken"):
            raise RuntimeError("player disappeared")
        return good

    controller = MprisController()
    monkeypatch.setattr(
        controller,
        "_list_player_names",
        AsyncMock(return_value=[
            "org.mpris.MediaPlayer2.broken",
            "org.mpris.MediaPlayer2.good",
        ]),
    )
    monkeypatch.setattr(controller, "_get_player_interface", fake_get_player)

    with caplog.at_level("WARNING"):
        await controller.pause()

    good.call_pause.assert_awaited_once()
    assert controller._paused == ["org.mpris.MediaPlayer2.good"]
    assert any("broken" in r.message for r in caplog.records)


@pytest.mark.asyncio
async def test_resume_logs_and_continues_when_single_player_fails(monkeypatch, caplog):
    from whisper_local.media._mpris import MprisController

    good = _make_player("Paused")

    async def fake_get_player(name: str):
        if name.endswith("broken"):
            raise RuntimeError("player disappeared")
        return good

    controller = MprisController()
    controller._paused = [
        "org.mpris.MediaPlayer2.broken",
        "org.mpris.MediaPlayer2.good",
    ]
    monkeypatch.setattr(controller, "_get_player_interface", fake_get_player)

    with caplog.at_level("WARNING"):
        await controller.resume()

    good.call_play.assert_awaited_once()
    assert controller._paused == []
    assert any("broken" in r.message for r in caplog.records)


@pytest.mark.asyncio
async def test_pause_is_noop_when_bus_unreachable(monkeypatch, caplog):
    from whisper_local.media._mpris import MprisController

    async def failing_connect(self):
        raise RuntimeError("no session bus")

    monkeypatch.setattr(
        "dbus_next.aio.MessageBus.connect", failing_connect, raising=False
    )
    controller = MprisController()

    with caplog.at_level("WARNING"):
        await controller.pause()

    assert controller._paused == []
    assert any("D-Bus" in r.message or "bus" in r.message.lower()
               for r in caplog.records)


@pytest.mark.asyncio
async def test_pause_skips_reconnect_after_bus_failure(monkeypatch):
    """Bus-Connect-Fehler darf nicht bei jedem Aufruf erneut versucht werden."""
    from whisper_local.media._mpris import MprisController

    connect_calls = 0

    async def failing_connect(self):
        nonlocal connect_calls
        connect_calls += 1
        raise RuntimeError("no session bus")

    monkeypatch.setattr(
        "dbus_next.aio.MessageBus.connect", failing_connect, raising=False
    )
    controller = MprisController()

    await controller.pause()
    await controller.pause()
    await controller.pause()

    assert connect_calls == 1