/
/
/
1"""Tests for the MPD player connection handling."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import Callable
7from typing import cast
8from unittest.mock import AsyncMock, MagicMock, patch
9
10from mpd import CommandError
11from mpd import ConnectionError as MPDConnectionError
12
13from music_assistant.providers.mpd.player import MPDPlayer
14
15
16def _mock_client() -> MagicMock:
17 """Return a mocked MPDClient that connects successfully."""
18 client = MagicMock()
19 client.connect = AsyncMock()
20 client.password = AsyncMock()
21 client.status = AsyncMock(return_value={"state": "stop"})
22 return client
23
24
25async def test_failed_authentication_closes_the_connection(
26 make_mpd_player: Callable[..., MPDPlayer],
27) -> None:
28 """A rejected password must not leave the connected client behind."""
29 player = make_mpd_player()
30 player.password = "wrong"
31 client = _mock_client()
32 client.password = AsyncMock(side_effect=CommandError("[3@0] {password} incorrect password"))
33
34 with patch("music_assistant.providers.mpd.player.MPDClient", return_value=client):
35 await player._connect()
36
37 client.disconnect.assert_called_once()
38 assert player._client is None
39 assert player.needs_setup is True
40
41
42async def test_failed_idle_connection_closes_the_command_connection(
43 make_mpd_player: Callable[..., MPDPlayer],
44) -> None:
45 """When the idle connection fails, the command connection must be closed too."""
46 player = make_mpd_player()
47 player.password = None
48 command_client = _mock_client()
49 idle_client = _mock_client()
50 idle_client.connect = AsyncMock(side_effect=MPDConnectionError("connection lost"))
51
52 with patch(
53 "music_assistant.providers.mpd.player.MPDClient",
54 side_effect=[command_client, idle_client],
55 ):
56 await player._connect()
57
58 command_client.disconnect.assert_called_once()
59 idle_client.disconnect.assert_called_once()
60 assert player._client is None
61 assert player._idle_client is None
62 assert player.available is False
63
64
65async def test_disconnect_cancels_a_pending_reconnect(
66 make_mpd_player: Callable[..., MPDPlayer],
67) -> None:
68 """Disconnecting must disarm a reconnect that is still waiting on its timer."""
69 player = make_mpd_player()
70 cast("MagicMock", player.mass.get_task).return_value = None
71
72 await player._disconnect()
73
74 cast("MagicMock", player.mass.cancel_timer).assert_called_once_with(player._reconnect_task_id)
75 cast("MagicMock", player.mass.cancel_task).assert_not_called()
76
77
78async def test_disconnect_stops_a_connect_attempt_in_flight(
79 make_mpd_player: Callable[..., MPDPlayer],
80) -> None:
81 """A connect attempt that already started must not outlive the teardown."""
82 player = make_mpd_player()
83 in_flight = asyncio.create_task(asyncio.sleep(30))
84 cast("MagicMock", player.mass.get_task).return_value = in_flight
85
86 await player._disconnect()
87
88 cast("MagicMock", player.mass.cancel_task).assert_called_once_with(player._reconnect_task_id)
89 in_flight.cancel()
90
91
92async def test_disconnect_never_cancels_its_own_connect_attempt(
93 make_mpd_player: Callable[..., MPDPlayer],
94) -> None:
95 """The teardown at the start of a connect attempt must not cancel that attempt."""
96 player = make_mpd_player()
97 cast("MagicMock", player.mass.get_task).return_value = asyncio.current_task()
98
99 await player._disconnect()
100
101 cast("MagicMock", player.mass.cancel_task).assert_not_called()
102