/
/
/
1"""Tests for Snapcast player liveness detection (abrupt power-off)."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import TYPE_CHECKING, cast
7
8from music_assistant.providers.snapcast.constants import SNAPCLIENT_STALE_THRESHOLD
9from music_assistant.providers.snapcast.player import SnapCastPlayer
10
11if TYPE_CHECKING:
12 from music_assistant.mass import MusicAssistant
13 from music_assistant.providers.snapcast.snap_cntrl_proto import SnapclientProto
14
15
16class _Clock:
17 """Controllable monotonic clock stand-in for loop.time()."""
18
19 def __init__(self) -> None:
20 self.now = 1000.0
21
22 def time(self) -> float:
23 """Return the current (fake) monotonic time."""
24 return self.now
25
26
27def _make_player(clock: _Clock, *, connected: bool, last_seen_sec: int) -> SnapCastPlayer:
28 player = SnapCastPlayer.__new__(SnapCastPlayer)
29 player._last_seen_raw = None
30 player._last_seen_changed = 0.0
31 player.mass = cast("MusicAssistant", SimpleNamespace(loop=clock))
32 player.snap_client = cast(
33 "SnapclientProto",
34 SimpleNamespace(
35 connected=connected,
36 _client={"lastSeen": {"sec": last_seen_sec, "usec": 0}},
37 ),
38 )
39 return player
40
41
42def test_alive_while_lastseen_advances() -> None:
43 """A connected client stays available as long as lastSeen keeps advancing."""
44 clock = _Clock()
45 player = _make_player(clock, connected=True, last_seen_sec=500)
46 assert player._is_alive() is True
47 for _ in range(5):
48 clock.now += SNAPCLIENT_STALE_THRESHOLD + 1
49 player.snap_client._client["lastSeen"]["sec"] += 1
50 assert player._is_alive() is True
51
52
53def test_becomes_unavailable_when_lastseen_frozen() -> None:
54 """A connected client whose lastSeen freezes flips to unavailable past the threshold."""
55 clock = _Clock()
56 player = _make_player(clock, connected=True, last_seen_sec=500)
57 assert player._is_alive() is True
58 clock.now += SNAPCLIENT_STALE_THRESHOLD - 1
59 assert player._is_alive() is True
60 clock.now += 2
61 assert player._is_alive() is False
62
63
64def test_disconnected_is_unavailable() -> None:
65 """A client reported disconnected by the server is unavailable regardless of lastSeen."""
66 clock = _Clock()
67 player = _make_player(clock, connected=False, last_seen_sec=500)
68 assert player._is_alive() is False
69
70
71def test_recovers_after_reconnect() -> None:
72 """A stale client becomes available again once lastSeen resumes advancing."""
73 clock = _Clock()
74 player = _make_player(clock, connected=True, last_seen_sec=500)
75 assert player._is_alive() is True
76 clock.now += SNAPCLIENT_STALE_THRESHOLD + 1
77 assert player._is_alive() is False
78 player.snap_client._client["lastSeen"]["sec"] += 1
79 assert player._is_alive() is True
80