/
/
/
1"""Unit tests for the Sonos S1 helpers."""
2
3from __future__ import annotations
4
5import inspect
6from typing import Any
7from unittest.mock import MagicMock
8
9import pytest
10from soco.exceptions import SoCoException, SoCoUPnPException
11
12from music_assistant.providers.sonos_s1.helpers import SonosUpdateError, soco_error
13from music_assistant.providers.sonos_s1.player import SonosPlayer
14
15IGNORED_ERROR_CODE = "501"
16OTHER_ERROR_CODE = "701"
17
18SOCO_ERRORS = [
19 pytest.param(SoCoException("boom"), id="soco"),
20 pytest.param(SoCoUPnPException("boom", OTHER_ERROR_CODE, b""), id="upnp"),
21 pytest.param(OSError("boom"), id="oserror"),
22 pytest.param(TimeoutError("boom"), id="timeout"),
23]
24
25SYNC_AND_ASYNC = ["probe", "regroup"]
26SYNC_AND_ASYNC_FILTERED = ["probe_filtered", "regroup_filtered"]
27
28
29class _Speaker(SonosPlayer):
30 """Minimal speaker exposing a decorated service call in a sync and an async flavour."""
31
32 def __init__(self, error: Exception | None = None) -> None:
33 soco = MagicMock()
34 soco._player_name = "Kitchen"
35 self.soco = soco
36 self._error = error
37
38 @soco_error()
39 def probe(self) -> str:
40 """Sync service call."""
41 return self._service_call()
42
43 @soco_error()
44 async def regroup(self) -> str:
45 """Async service call."""
46 return self._service_call()
47
48 @soco_error([IGNORED_ERROR_CODE])
49 def probe_filtered(self) -> str:
50 """Sync service call that tolerates one error code."""
51 return self._service_call()
52
53 @soco_error([IGNORED_ERROR_CODE])
54 async def regroup_filtered(self) -> str:
55 """Async service call that tolerates one error code."""
56 return self._service_call()
57
58 def _service_call(self) -> str:
59 if self._error:
60 raise self._error
61 return "ok"
62
63
64async def _invoke(speaker: _Speaker, method: str) -> Any:
65 """Call the named decorated method, awaiting it for the async flavour."""
66 result = getattr(speaker, method)()
67 return await result if inspect.isawaitable(result) else result
68
69
70@pytest.mark.parametrize("method", SYNC_AND_ASYNC)
71async def test_result_is_returned_untouched(method: str) -> None:
72 """A call that succeeds returns its own result."""
73 assert await _invoke(_Speaker(), method) == "ok"
74
75
76@pytest.mark.parametrize("error", SOCO_ERRORS)
77@pytest.mark.parametrize("method", SYNC_AND_ASYNC)
78async def test_soco_errors_become_a_sonos_update_error(method: str, error: Exception) -> None:
79 """Every kind of soco failure is reported as a SonosUpdateError naming the speaker."""
80 with pytest.raises(SonosUpdateError, match="Kitchen") as exc_info:
81 await _invoke(_Speaker(error), method)
82
83 assert exc_info.value.__cause__ is error
84
85
86@pytest.mark.parametrize("method", SYNC_AND_ASYNC_FILTERED)
87async def test_ignored_error_code_is_swallowed(method: str) -> None:
88 """An error code the caller opted to ignore yields None instead of an exception."""
89 error = SoCoUPnPException("boom", IGNORED_ERROR_CODE, b"")
90
91 assert await _invoke(_Speaker(error), method) is None
92
93
94@pytest.mark.parametrize("method", SYNC_AND_ASYNC_FILTERED)
95async def test_other_error_codes_are_still_raised(method: str) -> None:
96 """Error codes outside the ignore list are reported as usual."""
97 error = SoCoUPnPException("boom", OTHER_ERROR_CODE, b"")
98
99 with pytest.raises(SonosUpdateError, match="Kitchen"):
100 await _invoke(_Speaker(error), method)
101
102
103def test_async_method_stays_a_coroutine_function() -> None:
104 """Decorated coroutine methods remain awaitable for their callers."""
105 assert inspect.iscoroutinefunction(_Speaker.regroup)
106 assert not inspect.iscoroutinefunction(_Speaker.probe)
107