/
/
/
1"""Fixtures for Snapcast provider tests."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9
10
11class FakeSnapstream:
12 """Minimal Snapstream stand-in for testing ma_stream._register_tcp_server_source."""
13
14 def __init__(self, identifier: str, name: str, path: str = "") -> None:
15 """Initialize with stream identifier, name, and optional path."""
16 self.identifier = identifier
17 self.name = name
18 self.path = path
19 self._stream: dict[str, Any] = {"uri": {"host": "127.0.0.1"}}
20 self._callback = None
21
22 def set_callback(self, cb: Any) -> None:
23 """Register a callback for stream events."""
24 self._callback = cb
25
26
27class FakeSnapserver:
28 """Configurable mock implementing the subset of SnapserverProto used by ma_stream."""
29
30 def __init__(self) -> None:
31 """Initialize with empty stream registry and response queues."""
32 # streams added via stream_add_stream are kept here keyed by id
33 self._streams_by_id: dict[str, FakeSnapstream] = {}
34 # responses to return for each call to stream_add_stream, in order
35 # Each entry is a dict that looks like the snapserver result OR error payload.
36 self._add_stream_responses: list[dict[str, Any]] = []
37 # Calls captured for assertions
38 self.add_stream_calls: list[str] = []
39
40 self.stream_add_stream = AsyncMock(side_effect=self._add_stream_impl)
41 self.stream_remove_stream = AsyncMock()
42 self.status = AsyncMock(side_effect=self._status_impl)
43 self._status_payload: dict[str, Any] | None = None
44
45 @property
46 def streams(self) -> list[FakeSnapstream]:
47 """Return all registered streams."""
48 return list(self._streams_by_id.values())
49
50 def stream(self, identifier: str) -> FakeSnapstream:
51 """Return the stream with the given identifier."""
52 return self._streams_by_id[identifier]
53
54 def synchronize(self, status: dict[str, Any]) -> None:
55 """Populate _streams_by_id from a status payload, simulating real Snapserver."""
56 streams = status.get("server", {}).get("streams", [])
57 for s in streams:
58 sid = s.get("id")
59 if not sid:
60 continue
61 if sid not in self._streams_by_id:
62 self._streams_by_id[sid] = FakeSnapstream(sid, name=s.get("name", sid))
63
64 async def _status_impl(self) -> tuple[Any, Any]:
65 # status() in the real Snapserver is async and returns (result, error)
66 return (self._status_payload, None)
67
68 async def _add_stream_impl(self, stream_uri: str) -> dict[str, Any]:
69 self.add_stream_calls.append(stream_uri)
70 if self._add_stream_responses:
71 response = self._add_stream_responses.pop(0)
72 else:
73 response = {
74 "code": -32603,
75 "data": "Generic snapserver error",
76 "message": "Internal error",
77 }
78 # On a successful add, also register the stream in our internal store
79 if isinstance(response, dict) and "id" in response:
80 stream_id = response["id"]
81 # parse name= from uri for realism
82 name = stream_uri.split("&name=", 1)[1] if "&name=" in stream_uri else stream_id
83 self._streams_by_id[stream_id] = FakeSnapstream(stream_id, name, path=stream_uri)
84 return response
85
86 def queue_response(self, response: dict[str, Any]) -> None:
87 """Queue a response that the next stream_add_stream call will return."""
88 self._add_stream_responses.append(response)
89
90 def queue_success(self, stream_id: str = "stream-1") -> None:
91 """Queue a successful stream_add_stream response with the given stream id."""
92 self.queue_response({"id": stream_id})
93
94 def queue_name_collision(self) -> None:
95 """Queue a name-already-exists error response."""
96 self.queue_response(
97 {
98 "code": -32603,
99 "data": 'Stream with name "x" already exists',
100 "message": "Internal error",
101 }
102 )
103
104 def queue_other_error(self, data: str = "bind: Address already in use") -> None:
105 """Queue a generic (non-name-collision) error response."""
106 self.queue_response({"code": -32603, "data": data, "message": "Internal error"})
107
108 def cache_stream_directly(self, stream_id: str, name: str) -> FakeSnapstream:
109 """
110 Pre-register a stream in the local cache (skipping the status round-trip).
111
112 Use this when the test wants the orphan to be visible WITHOUT requiring
113 adopt() to call status() + synchronize() first.
114 """
115 s = FakeSnapstream(stream_id, name)
116 self._streams_by_id[stream_id] = s
117 return s
118
119 def stage_orphan_stream(self, stream_id: str, name: str) -> None:
120 """
121 Stage an orphan stream that only appears after .synchronize(.status()).
122
123 Use this when the test wants to verify that adopt() falls back to a
124 status-driven resync (the actual Bug B post-MA-restart scenario).
125 The stream is NOT in _streams_by_id until synchronize() is called.
126 """
127 self._status_payload = {
128 "server": {
129 "streams": [{"id": stream_id, "name": name}],
130 "groups": [],
131 }
132 }
133
134
135@pytest.fixture
136def fake_snapserver() -> FakeSnapserver:
137 """Provide a fresh FakeSnapserver for each test."""
138 return FakeSnapserver()
139
140
141@pytest.fixture
142def fake_provider(fake_snapserver: FakeSnapserver) -> MagicMock:
143 """Provide a minimal SnapCastProvider mock wired to fake_snapserver."""
144 provider = MagicMock()
145 provider._snapserver = fake_snapserver
146 provider._snapcast_stream_idle_threshold = 60000
147 provider._use_builtin_server = False
148 provider.logger = MagicMock()
149 return provider
150