/
/
/
1"""
2Tests for surfacing receiver media errors from ChromecastPlayer.
3
4Regression tests for https://github.com/music-assistant/support/issues/5981, where a
5receiver answered every LOAD with LOAD_FAILED and a media status carrying
6``idleReason: ERROR``, and Music Assistant logged nothing: the player silently
7returned to idle. A media error reported by the receiver must be visible in the log.
8
9The LOAD_FAILED message itself only reaches the ``load_media_failed`` listener when
10the receiver includes a detailed error code, so the media status path is the one
11that must catch the general case.
12"""
13
14from __future__ import annotations
15
16from typing import Any, cast
17from unittest.mock import MagicMock
18
19from music_assistant_models.enums import PlaybackState
20
21from music_assistant.providers.chromecast.constants import MASS_APP_ID
22from music_assistant.providers.chromecast.player import ChromecastPlayer
23
24
25def _fake_player(*, flow_underrun: bool = False) -> Any:
26 """
27 Build an idle, ungrouped Cast player on mocked collaborators.
28
29 :param flow_underrun: Whether the queue's flow stream has been fully consumed.
30 """
31 # __init__ is skipped: it needs a provider, cast info and a live Chromecast connection.
32 # Typed as Any because the collaborators below are read back as the mocks they are.
33 fake = cast("Any", ChromecastPlayer.__new__(ChromecastPlayer))
34 fake.mass = MagicMock()
35 fake.logger = MagicMock()
36 fake.cc = MagicMock(app_id=MASS_APP_ID)
37 fake.update_state = MagicMock()
38 fake._flow_stream_underrun = MagicMock(return_value=flow_underrun)
39 fake._media_error_reported = False
40 fake._app_quit_task_id = "cast_quit_app_test"
41 fake.active_cast_group = None
42 # display_name is a cached property, normally built from the config in __init__
43 fake._cache = {"display_name": "Test Cast"}
44 # playback_state is a read-only property, so it is seeded through its backing field
45 fake._attr_playback_state = PlaybackState.IDLE
46 return fake
47
48
49def _error_status() -> MagicMock:
50 status = MagicMock()
51 status.content_id = "http://192.168.1.58:8097/flow/abc/track.flac"
52 status.player_is_playing = False
53 status.player_is_paused = False
54 status.player_is_idle = True
55 status.idle_reason = "ERROR"
56 return status
57
58
59def _playing_status() -> MagicMock:
60 status = MagicMock()
61 status.content_id = "http://192.168.1.58:8097/flow/abc/track.flac"
62 status.player_is_playing = True
63 status.player_is_paused = False
64 status.player_is_idle = False
65 status.idle_reason = None
66 return status
67
68
69def test_idle_error_is_logged() -> None:
70 """A media status with idleReason ERROR produces a warning naming the media."""
71 fake = _fake_player()
72
73 fake._handle_media_status(_error_status())
74
75 fake.logger.warning.assert_called_once()
76 assert "track.flac" in str(fake.logger.warning.call_args)
77
78
79def test_repeated_error_status_is_logged_once() -> None:
80 """The receiver echoes the error status several times; only the first is logged."""
81 fake = _fake_player()
82
83 fake._handle_media_status(_error_status())
84 fake._handle_media_status(_error_status())
85 fake._handle_media_status(_error_status())
86
87 fake.logger.warning.assert_called_once()
88
89
90def test_error_logged_again_after_recovery() -> None:
91 """A new error after successful playback is a new incident and is logged again."""
92 fake = _fake_player()
93
94 fake._handle_media_status(_error_status())
95 fake._handle_media_status(_playing_status())
96 fake._handle_media_status(_error_status())
97
98 assert fake.logger.warning.call_count == 2
99
100
101def test_group_error_is_not_logged_by_every_member() -> None:
102 """A group's error reaches all its members, but only the group itself reports it."""
103 group = MagicMock()
104 # a plain spec= mock has no 'cc', which is set in __init__
105 group.__class__ = ChromecastPlayer # type: ignore[assignment]
106 group.cc.media_controller.status = _error_status()
107 member = _fake_player()
108 member.active_cast_group = "group-uuid"
109 member.mass.players.get_player = MagicMock(return_value=group)
110
111 member._handle_media_status(_error_status())
112
113 member.logger.warning.assert_not_called()
114 # assert the group status was really processed, so the check above cannot pass
115 # just because the member bailed out before reaching the error handling
116 member.mass.players.get_player.assert_called_once_with("group-uuid")
117 member.update_state.assert_called_once()
118
119
120def test_flow_stream_underrun_is_not_an_error() -> None:
121 """An idle ERROR at the end of a fully consumed flow stream is expected, not logged."""
122 fake = _fake_player(flow_underrun=True)
123
124 fake._handle_media_status(_error_status())
125
126 fake.logger.warning.assert_not_called()
127
128
129def test_load_media_failed_logs_the_error_code() -> None:
130 """A LOAD_FAILED with a detailed error code is logged with its meaning."""
131 fake = _fake_player()
132
133 fake._handle_load_media_failed(1, 104)
134
135 fake.logger.warning.assert_called_once()
136 assert "104" in str(fake.logger.warning.call_args)
137 assert fake._media_error_reported is True
138