/
/
/
1"""Tests for the Sendspin management-call timeout and transport-error mapping."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING, cast
7
8import pytest
9
10import music_assistant.providers.sendspin.provider as provider_module
11from music_assistant.providers.sendspin.helpers import SecurityActionError
12from music_assistant.providers.sendspin.provider import SendspinProvider
13
14if TYPE_CHECKING:
15 from aiosendspin.server.connection import SendspinConnection
16
17
18class _FakeConnection:
19 """Connection stand-in recording whether disconnect was awaited."""
20
21 def __init__(self) -> None:
22 self.disconnect_calls = 0
23
24 async def disconnect(self) -> None:
25 self.disconnect_calls += 1
26
27
28async def test_management_call_returns_result() -> None:
29 """A prompt reply passes straight through, leaving the connection untouched."""
30 provider = SendspinProvider.__new__(SendspinProvider)
31 conn = _FakeConnection()
32
33 async def _reply() -> str:
34 await asyncio.sleep(0)
35 return "ok"
36
37 result = await provider._management_call(cast("SendspinConnection", conn), _reply())
38 assert result == "ok"
39 assert conn.disconnect_calls == 0
40
41
42async def test_management_call_timeout_disconnects(monkeypatch: pytest.MonkeyPatch) -> None:
43 """A timed-out request drops the connection so its ordered channel can't desync the next."""
44 monkeypatch.setattr(provider_module, "MANAGEMENT_REQUEST_TIMEOUT", 0.01)
45 provider = SendspinProvider.__new__(SendspinProvider)
46 conn = _FakeConnection()
47
48 async def _hang() -> str:
49 await asyncio.Event().wait()
50 return ""
51
52 with pytest.raises(SecurityActionError) as excinfo:
53 await provider._management_call(cast("SendspinConnection", conn), _hang())
54 assert excinfo.value.alert_key == "management_error_timeout"
55 assert conn.disconnect_calls == 1
56
57
58async def test_management_call_runtime_error_maps_without_disconnect() -> None:
59 """A transport RuntimeError becomes a SecurityActionError and leaves the connection alone."""
60 provider = SendspinProvider.__new__(SendspinProvider)
61 conn = _FakeConnection()
62
63 async def _boom() -> str:
64 await asyncio.sleep(0)
65 raise RuntimeError("connection is not active")
66
67 with pytest.raises(SecurityActionError) as excinfo:
68 await provider._management_call(cast("SendspinConnection", conn), _boom())
69 assert excinfo.value.alert_key == "management_error_generic"
70 assert excinfo.value.detail == "connection is not active"
71 assert conn.disconnect_calls == 0
72