/
/
/
1"""Tests for the Spotify Soloist shared helpers (binary manager + WebSocket client)."""
2
3from __future__ import annotations
4
5import asyncio
6import hashlib
7import io
8import json
9import logging
10import platform
11import tarfile
12import time
13from collections.abc import AsyncGenerator, Callable
14from datetime import UTC, datetime
15from pathlib import Path
16from types import SimpleNamespace
17from typing import TYPE_CHECKING, Any, Self, cast
18
19import pytest
20from aiohttp import ClientError, WSMessage, WSMsgType
21
22from music_assistant.providers.spotify_connect.soloist import runtime as soloist
23from music_assistant.providers.spotify_connect.soloist.runtime import (
24 BuildExpiredError,
25 ConsentRequiredError,
26 DownloadFailedError,
27 InvalidArchiveError,
28 SoloistAuthState,
29 SoloistBinaryManager,
30 SoloistClient,
31 SoloistError,
32 SoloistEvent,
33 SoloistPlaybackState,
34 SoloistPositionSync,
35 SoloistQueueChanged,
36 SoloistTrackChanged,
37 SoloistVolumeChanged,
38 UnsupportedPlatformError,
39)
40
41if TYPE_CHECKING:
42 from music_assistant.mass import MusicAssistant
43
44_CDN_URL = "https://soloist-builds.spotifycdn.com/soloist_release_{arch}.tar.gz"
45# build timestamp relative to now so the fake build never ages past the 90-day expiry
46_VERSION_OUTPUT = (
47 f"soloist version 1.2.3\nbuild {datetime.now(tz=UTC):%Y-%m-%dT%H:%M:%SZ} linux/x86_64"
48).encode()
49
50
51def _elf_binary(arch: str, marker: bytes = b"GOOD") -> bytes:
52 """Return a minimal (fake) ELF executable for the given soloist architecture."""
53 machine = {"arm64": 0xB7, "arm32": 0x28, "x86_64": 0x3E}[arch]
54 header = bytearray(20)
55 header[0:4] = b"\x7fELF"
56 header[4] = 1 if arch == "arm32" else 2 # EI_CLASS
57 header[5] = 1 # EI_DATA: little-endian
58 header[6] = 1 # EI_VERSION
59 header[16:18] = (2).to_bytes(2, "little") # e_type: ET_EXEC
60 header[18:20] = machine.to_bytes(2, "little")
61 return bytes(header) + marker + b"\x00" * 64
62
63
64def _build_archive(
65 path: Path,
66 files: dict[str, bytes] | None = None,
67 *,
68 symlink: tuple[str, str] | None = None,
69) -> bytes:
70 """Build a tar.gz archive with the given members and return its raw bytes."""
71 with tarfile.open(path, "w:gz") as tar:
72 for name, content in (files or {}).items():
73 info = tarfile.TarInfo(name)
74 info.size = len(content)
75 tar.addfile(info, io.BytesIO(content))
76 if symlink is not None:
77 info = tarfile.TarInfo(symlink[0])
78 info.type = tarfile.SYMTYPE
79 info.linkname = symlink[1]
80 tar.addfile(info)
81 return path.read_bytes()
82
83
84class _FakeContent:
85 """Response body that hands out chunks like aiohttp's StreamReader."""
86
87 def __init__(self, body: bytes) -> None:
88 self._body = body
89
90 async def iter_chunked(self, n: int) -> AsyncGenerator[bytes]:
91 """Yield the body in chunks of at most n bytes."""
92 for i in range(0, len(self._body), n):
93 await asyncio.sleep(0)
94 yield self._body[i : i + n]
95
96
97class _FakeResponse:
98 """Stand-in for an aiohttp response context manager."""
99
100 def __init__(
101 self, status: int = 200, headers: dict[str, str] | None = None, body: bytes = b""
102 ) -> None:
103 self.status = status
104 self.headers = headers or {}
105 self.content = _FakeContent(body)
106
107 async def __aenter__(self) -> Self:
108 return self
109
110 async def __aexit__(self, *exc_info: object) -> None:
111 return None
112
113
114_Handler = Callable[[str, str], _FakeResponse]
115
116
117class _FakeSession:
118 """Fake aiohttp session that records every request it receives."""
119
120 def __init__(self, handler: _Handler) -> None:
121 self.handler = handler
122 self.requests: list[tuple[str, str]] = []
123
124 def get(self, url: str, **_kwargs: Any) -> _FakeResponse:
125 """Issue a fake GET request."""
126 return self._request("GET", url)
127
128 def head(self, url: str, **_kwargs: Any) -> _FakeResponse:
129 """Issue a fake HEAD request."""
130 return self._request("HEAD", url)
131
132 def _request(self, method: str, url: str) -> _FakeResponse:
133 self.requests.append((method, url))
134 return self.handler(method, url)
135
136
137def _serve_archive(archive: bytes, etag: str = '"v1"') -> _Handler:
138 """Return a request handler that serves the given archive bytes for any URL."""
139
140 def handler(method: str, _url: str) -> _FakeResponse:
141 if method == "HEAD":
142 return _FakeResponse(headers={"ETag": etag})
143 return _FakeResponse(headers={"ETag": etag}, body=archive)
144
145 return handler
146
147
148def _offline(_method: str, _url: str) -> _FakeResponse:
149 """Request handler that behaves as if there is no network at all."""
150 raise ClientError("no route to host")
151
152
153def _make_manager(tmp_path: Path, handler: _Handler) -> tuple[SoloistBinaryManager, _FakeSession]:
154 """Create a binary manager on a fake mass with the given request handler."""
155 session = _FakeSession(handler)
156 mass = SimpleNamespace(storage_path=str(tmp_path / "storage"), http_session=session)
157 return SoloistBinaryManager(cast("MusicAssistant", mass)), session
158
159
160def _install_dir(tmp_path: Path) -> Path:
161 """Return the manager's install directory for the given tmp_path."""
162 return tmp_path / "storage" / "soloist"
163
164
165def _age_metadata(tmp_path: Path, days: float = 80.0) -> None:
166 """
167 Rewrite the persisted install metadata as if the build were days old.
168
169 The default lands inside the update window (76 days) while staying short of
170 the hard 90-day expiry, so the install counts as refreshable-but-valid.
171 """
172 meta_path = _install_dir(tmp_path) / "soloist.meta.json"
173 meta = json.loads(meta_path.read_text(encoding="utf-8"))
174 aged = time.time() - days * 86400
175 meta["installed_at"] = aged
176 meta["build_timestamp"] = aged
177 meta_path.write_text(json.dumps(meta), encoding="utf-8")
178
179
180async def _fake_check_output(*args: str, **_kwargs: Any) -> tuple[int, bytes]:
181 """Fake --version subprocess call whose result depends on markers in the binary."""
182 content = Path(args[0]).read_bytes()
183 if b"EXPIRED" in content:
184 return (10, b"soloist build expired")
185 if b"BROKEN" in content:
186 return (1, b"crash")
187 return (0, _VERSION_OUTPUT)
188
189
190@pytest.fixture
191def fake_version_cmd(monkeypatch: pytest.MonkeyPatch) -> None:
192 """Replace the --version subprocess call with a marker-based fake."""
193 monkeypatch.setattr(soloist, "check_output", _fake_check_output)
194
195
196@pytest.fixture(autouse=True)
197def _reset_verify_cache(monkeypatch: pytest.MonkeyPatch) -> None:
198 """Isolate the module-level recent-verification stamp between tests."""
199 monkeypatch.setattr(soloist, "_last_verified", None)
200
201
202@pytest.fixture
203def linux_platform(monkeypatch: pytest.MonkeyPatch) -> None:
204 """Pretend to run on Linux x86_64 (tests run on macOS)."""
205 monkeypatch.setattr(platform, "system", lambda: "Linux")
206 monkeypatch.setattr(platform, "machine", lambda: "x86_64")
207
208
209@pytest.mark.usefixtures("fake_version_cmd")
210@pytest.mark.parametrize(
211 ("machine", "arch"),
212 [
213 ("aarch64", "arm64"),
214 ("arm64", "arm64"),
215 ("armv7l", "arm32"),
216 ("armv8l", "arm32"),
217 ("x86_64", "x86_64"),
218 ("amd64", "x86_64"),
219 ],
220)
221async def test_arch_maps_to_cdn_artifact(
222 machine: str, arch: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
223) -> None:
224 """Each supported machine downloads the matching CDN artifact."""
225 monkeypatch.setattr(platform, "system", lambda: "Linux")
226 monkeypatch.setattr(platform, "machine", lambda: machine)
227 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary(arch)})
228 manager, session = _make_manager(tmp_path, _serve_archive(archive))
229
230 path = await manager.ensure_binary(consent=True)
231
232 assert path.is_file()
233 assert session.requests == [("GET", _CDN_URL.format(arch=arch))]
234
235
236async def test_non_linux_platform_rejected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
237 """A non-Linux platform is rejected without any network access."""
238 monkeypatch.setattr(platform, "system", lambda: "Darwin")
239 monkeypatch.setattr(platform, "machine", lambda: "arm64")
240 manager, session = _make_manager(tmp_path, _offline)
241
242 with pytest.raises(UnsupportedPlatformError):
243 await manager.ensure_binary(consent=True)
244 assert session.requests == []
245
246
247async def test_unknown_machine_rejected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
248 """An unknown machine architecture is rejected without any network access."""
249 monkeypatch.setattr(platform, "system", lambda: "Linux")
250 monkeypatch.setattr(platform, "machine", lambda: "mips64")
251 manager, session = _make_manager(tmp_path, _offline)
252
253 with pytest.raises(UnsupportedPlatformError):
254 await manager.ensure_binary(consent=True)
255 assert session.requests == []
256
257
258@pytest.mark.usefixtures("linux_platform")
259async def test_download_requires_consent(tmp_path: Path) -> None:
260 """Without consent no download is attempted and no network call is made."""
261 manager, session = _make_manager(tmp_path, _offline)
262
263 with pytest.raises(ConsentRequiredError):
264 await manager.ensure_binary(consent=False)
265 assert session.requests == []
266
267
268@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
269async def test_installed_binary_returned_without_network(tmp_path: Path) -> None:
270 """An already-installed valid binary is returned without consent or network."""
271 install_dir = _install_dir(tmp_path)
272 install_dir.mkdir(parents=True)
273 (install_dir / "soloist").write_bytes(_elf_binary("x86_64"))
274 manager, session = _make_manager(tmp_path, _offline)
275
276 path = await manager.ensure_binary(consent=False)
277
278 assert path == install_dir / "soloist"
279 assert session.requests == []
280
281
282@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
283@pytest.mark.parametrize("redirect_host", ["evil.example.com", "evilspotifycdn.com"])
284async def test_redirect_outside_allowlist_rejected(tmp_path: Path, redirect_host: str) -> None:
285 """A redirect to a host outside Spotify's infrastructure aborts the download."""
286
287 def handler(_method: str, _url: str) -> _FakeResponse:
288 return _FakeResponse(
289 status=302, headers={"Location": f"https://{redirect_host}/soloist.tar.gz"}
290 )
291
292 manager, session = _make_manager(tmp_path, handler)
293
294 with pytest.raises(DownloadFailedError, match="untrusted host"):
295 await manager.ensure_binary(consent=True)
296 # only the initial request went out, the redirect was never followed
297 assert len(session.requests) == 1
298
299
300@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
301async def test_redirect_within_allowlist_followed(tmp_path: Path) -> None:
302 """A redirect within Spotify's infrastructure is followed and the download succeeds."""
303 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
304 redirect_url = "https://downloads.spotify.com/soloist_release_x86_64.tar.gz"
305
306 def handler(_method: str, url: str) -> _FakeResponse:
307 if url != redirect_url:
308 return _FakeResponse(status=302, headers={"Location": redirect_url})
309 return _FakeResponse(body=archive)
310
311 manager, session = _make_manager(tmp_path, handler)
312
313 path = await manager.ensure_binary(consent=True)
314
315 assert path.is_file()
316 assert session.requests[-1] == ("GET", redirect_url)
317
318
319@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
320@pytest.mark.parametrize(
321 "files",
322 [
323 {"../soloist": b"payload"}, # path traversal
324 {"/soloist": b"payload"}, # absolute path
325 {"README": b"docs"}, # no soloist binary at all
326 ],
327)
328async def test_unsafe_or_unexpected_archive_rejected(
329 tmp_path: Path, files: dict[str, bytes]
330) -> None:
331 """Archives with traversal, absolute paths or no binary at all are rejected."""
332 archive = _build_archive(tmp_path / "a.tar.gz", files)
333 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
334
335 with pytest.raises(InvalidArchiveError):
336 await manager.ensure_binary(consent=True)
337
338
339@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
340async def test_archive_siblings_are_ignored(tmp_path: Path) -> None:
341 """The release ships docs beside the binary: they are skipped, not rejected."""
342 archive = _build_archive(
343 tmp_path / "a.tar.gz",
344 {
345 "CHANGELOG.md": b"# changelog",
346 "THIRD_PARTY_LICENSES.txt": b"licenses",
347 "soloist": _elf_binary("x86_64"),
348 },
349 )
350 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
351
352 path = await manager.ensure_binary(consent=True)
353
354 assert path.is_file()
355 assert path.read_bytes() == _elf_binary("x86_64")
356 assert not (path.parent / "CHANGELOG.md").exists()
357
358
359@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
360async def test_symlink_archive_rejected(tmp_path: Path) -> None:
361 """An archive delivering soloist as a symlink is rejected."""
362 archive = _build_archive(tmp_path / "a.tar.gz", symlink=("soloist", "/etc/passwd"))
363 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
364
365 with pytest.raises(InvalidArchiveError):
366 await manager.ensure_binary(consent=True)
367
368
369@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
370async def test_garbage_archive_rejected(tmp_path: Path) -> None:
371 """A response that is not a tar.gz archive at all is rejected."""
372 manager, _ = _make_manager(tmp_path, _serve_archive(b"this is not a tarball"))
373
374 with pytest.raises(InvalidArchiveError):
375 await manager.ensure_binary(consent=True)
376
377
378@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
379@pytest.mark.parametrize("content", [_elf_binary("arm64"), b"#!/bin/sh\necho not an elf\n"])
380async def test_wrong_or_non_elf_binary_rejected(tmp_path: Path, content: bytes) -> None:
381 """A binary for another architecture (or not an ELF at all) is rejected."""
382 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": content})
383 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
384
385 with pytest.raises(InvalidArchiveError):
386 await manager.ensure_binary(consent=True)
387 assert not manager.binary_path.exists()
388
389
390@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
391async def test_failed_validation_leaves_no_binary(tmp_path: Path) -> None:
392 """A fresh install whose binary fails --version validation leaves nothing behind."""
393 archive = _build_archive(
394 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"BROKEN")}
395 )
396 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
397
398 with pytest.raises(InvalidArchiveError):
399 await manager.ensure_binary(consent=True)
400 assert not manager.binary_path.exists()
401
402
403@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
404async def test_fresh_download_of_expired_build(tmp_path: Path) -> None:
405 """A freshly downloaded build that reports exit code 10 raises BuildExpiredError."""
406 archive = _build_archive(
407 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"EXPIRED")}
408 )
409 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
410
411 with pytest.raises(BuildExpiredError):
412 await manager.ensure_binary(consent=True)
413 assert not manager.binary_path.exists()
414
415
416@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
417async def test_rollback_restores_previous_binary(
418 tmp_path: Path, caplog: pytest.LogCaptureFixture
419) -> None:
420 """A failed replacement is rolled back to the previously installed binary."""
421 good = _build_archive(
422 tmp_path / "good.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
423 )
424 broken = _build_archive(
425 tmp_path / "broken.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"BROKEN")}
426 )
427 manager, session = _make_manager(tmp_path, _serve_archive(good, etag='"v1"'))
428 await manager.ensure_binary(consent=True)
429 _age_metadata(tmp_path)
430 session.handler = _serve_archive(broken, etag='"v2"')
431
432 with caplog.at_level(logging.WARNING):
433 path = await manager.ensure_fresh(consent=True)
434
435 assert path == manager.binary_path
436 assert b"GOOD-BUILD-A" in path.read_bytes()
437 assert not (_install_dir(tmp_path) / "soloist.prev").exists()
438 assert manager.diagnostics()["etag"] == '"v1"'
439 assert "keeping the current binary" in caplog.text
440
441
442@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
443async def test_refresh_installs_new_build(tmp_path: Path) -> None:
444 """An aged install is replaced when the CDN serves a different build."""
445 build_a = _build_archive(
446 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
447 )
448 build_b = _build_archive(
449 tmp_path / "b.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-B")}
450 )
451 manager, session = _make_manager(tmp_path, _serve_archive(build_a, etag='"v1"'))
452 await manager.ensure_binary(consent=True)
453 _age_metadata(tmp_path)
454 session.handler = _serve_archive(build_b, etag='"v2"')
455
456 path = await manager.ensure_fresh(consent=True)
457
458 assert b"GOOD-BUILD-B" in path.read_bytes()
459 diag = manager.diagnostics()
460 assert diag["etag"] == '"v2"'
461 assert diag["sha256"] == hashlib.sha256(build_b).hexdigest()
462
463
464@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
465async def test_refresh_skipped_when_etag_unchanged(tmp_path: Path) -> None:
466 """An aged install is kept when the CDN still serves the same build."""
467 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
468 manager, session = _make_manager(tmp_path, _serve_archive(archive, etag='"v1"'))
469 await manager.ensure_binary(consent=True)
470 _age_metadata(tmp_path)
471 session.requests.clear()
472
473 await manager.ensure_fresh(consent=True)
474
475 assert session.requests == [("HEAD", _CDN_URL.format(arch="x86_64"))]
476
477
478@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
479async def test_offline_refresh_returns_valid_binary(
480 tmp_path: Path, caplog: pytest.LogCaptureFixture
481) -> None:
482 """When offline, a still-valid installed binary is returned with a warning."""
483 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
484 manager, session = _make_manager(tmp_path, _serve_archive(archive))
485 await manager.ensure_binary(consent=True)
486 _age_metadata(tmp_path)
487 session.handler = _offline
488
489 with caplog.at_level(logging.WARNING):
490 path = await manager.ensure_fresh(consent=True)
491
492 assert path == manager.binary_path
493 assert "Unable to check for a soloist update" in caplog.text
494
495
496@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
497async def test_offline_with_expired_binary_raises(tmp_path: Path) -> None:
498 """When offline and the installed build already expired, BuildExpiredError is raised."""
499 install_dir = _install_dir(tmp_path)
500 install_dir.mkdir(parents=True)
501 (install_dir / "soloist").write_bytes(_elf_binary("x86_64", marker=b"EXPIRED"))
502 manager, _ = _make_manager(tmp_path, _offline)
503
504 with pytest.raises(BuildExpiredError):
505 await manager.ensure_fresh(consent=True)
506
507
508@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
509async def test_recent_verification_shared_across_managers(
510 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
511) -> None:
512 """Back-to-back ensure_fresh calls run the --version/CDN verification only once."""
513 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
514 manager1, session1 = _make_manager(tmp_path, _serve_archive(archive))
515 await manager1.ensure_binary(consent=True)
516 _age_metadata(tmp_path) # old enough that a full verification also HEADs the CDN
517 version_calls: list[str] = []
518
519 async def _counting_check_output(*args: str, **kwargs: Any) -> tuple[int, bytes]:
520 version_calls.append(args[0])
521 return await _fake_check_output(*args, **kwargs)
522
523 monkeypatch.setattr(soloist, "check_output", _counting_check_output)
524 session1.requests.clear()
525 manager2, session2 = _make_manager(tmp_path, _serve_archive(archive))
526
527 path1 = await manager1.ensure_fresh(consent=True)
528 path2 = await manager2.ensure_fresh(consent=True)
529
530 assert path1 == path2 == manager1.binary_path
531 # the second manager reuses the just-completed verification entirely
532 assert len(version_calls) == 1
533 assert session1.requests == [("HEAD", _CDN_URL.format(arch="x86_64"))]
534 assert session2.requests == []
535
536
537@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
538async def test_concurrent_callers_share_one_download(tmp_path: Path) -> None:
539 """Concurrent ensure_binary callers trigger exactly one download."""
540 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
541 manager, session = _make_manager(tmp_path, _serve_archive(archive))
542
543 paths = await asyncio.gather(*(manager.ensure_binary(consent=True) for _ in range(5)))
544
545 assert all(path == manager.binary_path for path in paths)
546 assert [req for req in session.requests if req[0] == "GET"] == [
547 ("GET", _CDN_URL.format(arch="x86_64"))
548 ]
549
550
551@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
552async def test_diagnostics_contains_no_secrets(tmp_path: Path) -> None:
553 """Diagnostics exposes install/build metadata only, never any key material."""
554 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
555 manager, _ = _make_manager(tmp_path, _serve_archive(archive, etag='"v1"'))
556 assert manager.diagnostics() == {"installed": False}
557
558 await manager.ensure_binary(consent=True)
559 diag = manager.diagnostics()
560
561 assert set(diag) == {
562 "installed",
563 "sha256",
564 "etag",
565 "version",
566 "version_raw",
567 "installed_at",
568 "build_timestamp",
569 "expires_at",
570 }
571 assert diag["installed"] is True
572 assert diag["sha256"] == hashlib.sha256(archive).hexdigest()
573 assert diag["etag"] == '"v1"'
574 assert diag["version"] == "1.2.3"
575 assert diag["expires_at"] == pytest.approx(diag["build_timestamp"] + 90 * 24 * 3600)
576
577
578class _FakeWebSocket:
579 """Fake events WebSocket: an async iterator fed from a queue, recording sent frames."""
580
581 def __init__(self) -> None:
582 self.queue: asyncio.Queue[WSMessage | None] = asyncio.Queue()
583 self.sent: list[dict[str, Any]] = []
584 self.closed = False
585
586 async def __aenter__(self) -> Self:
587 return self
588
589 async def __aexit__(self, *exc_info: object) -> None:
590 self.closed = True
591
592 def __aiter__(self) -> _FakeWebSocket:
593 return self
594
595 async def __anext__(self) -> WSMessage:
596 msg = await self.queue.get()
597 if msg is None:
598 raise StopAsyncIteration
599 return msg
600
601 async def send_json(self, data: dict[str, Any]) -> None:
602 """Record an outgoing JSON frame."""
603 self.sent.append(data)
604
605 def exception(self) -> BaseException | None:
606 """Return the connection error (never set for this fake)."""
607 return None
608
609
610def _make_client(data_dir: Path, ws: _FakeWebSocket) -> SoloistClient:
611 """Create a client for the given data dir whose session connects to the fake ws."""
612 mass = SimpleNamespace(http_session=SimpleNamespace(ws_connect=lambda *_a, **_kw: ws))
613 return SoloistClient(cast("MusicAssistant", mass), data_dir, logging.getLogger("test.soloist"))
614
615
616def _publish_endpoint(data_dir: Path, addr: str = "127.0.0.1", port: str = "8765") -> None:
617 """Write the ws.addr/ws.port endpoint files like the daemon does."""
618 (data_dir / "ws.addr").write_text(f"{addr}\n", encoding="utf-8")
619 (data_dir / "ws.port").write_text(f"{port}\n", encoding="utf-8")
620
621
622def _text_msg(payload: dict[str, Any]) -> WSMessage:
623 """Wrap an event payload in a WebSocket TEXT message."""
624 return WSMessage(WSMsgType.TEXT, json.dumps(payload), None)
625
626
627async def test_endpoint_discovery_polls_until_ready(
628 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
629) -> None:
630 """Endpoint discovery keeps polling until both files exist and parse."""
631 monkeypatch.setattr(soloist, "_ENDPOINT_POLL_INTERVAL", 0.01)
632 client = _make_client(tmp_path, _FakeWebSocket())
633 task = asyncio.create_task(client.wait_until_ready(timeout=5.0))
634
635 await asyncio.sleep(0.05)
636 assert not task.done()
637 (tmp_path / "ws.addr").write_text("127.0.0.1\n", encoding="utf-8")
638 (tmp_path / "ws.port").write_text("not-a-port\n", encoding="utf-8")
639 await asyncio.sleep(0.05)
640 assert not task.done() # unparsable port: keep polling
641 (tmp_path / "ws.port").write_text("8765\n", encoding="utf-8")
642
643 assert await task is True
644
645
646async def test_endpoint_discovery_times_out(
647 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
648) -> None:
649 """Endpoint discovery returns False when the files never appear."""
650 monkeypatch.setattr(soloist, "_ENDPOINT_POLL_INTERVAL", 0.01)
651 client = _make_client(tmp_path, _FakeWebSocket())
652
653 assert await client.wait_until_ready(timeout=0.05) is False
654
655
656async def test_event_dispatch_decodes_documented_payloads(tmp_path: Path) -> None:
657 """Documented event payloads are decoded into their typed models."""
658 _publish_endpoint(tmp_path)
659 ws = _FakeWebSocket()
660 client = _make_client(tmp_path, ws)
661 playback_state = {
662 "type": "playback_state",
663 "status": "playing",
664 "item": {
665 "uri": "spotify:track:2JRo0gjbX4GrCqBYdRohoo",
666 "entity_type": "track",
667 "decorations": {
668 "identity": {"name": "My Song"},
669 "playback": {"duration_ms": 210000, "content_ratings": []},
670 },
671 },
672 "context": {
673 "uri": "spotify:playlist:37i9dQZF1DXcBWIGoYBM5M",
674 "entity_type": "playlist",
675 "decorations": {"identity": {"name": "Today's Top Hits"}},
676 },
677 "position": {"position_ms": 45000, "timestamp_ms": 1747654321000, "speed": 1.0},
678 "volume": 65,
679 "is_active": True,
680 "options": {"shuffle": False, "repeat": "off", "playback_speed": 1.0, "modes": {}},
681 "available_actions": {"pause": {}, "seek_forward": {"step_ms": 15000}},
682 }
683 for payload in (
684 {"type": "auth_state", "logged_in": True, "is_active": True, "device_name": "Kitchen"},
685 playback_state,
686 {"type": "volume_changed", "volume": 42},
687 {
688 "type": "position_sync",
689 "position": {"position_ms": 45000, "timestamp_ms": 1747654321000, "speed": 1.0},
690 },
691 {
692 "type": "queue_changed",
693 "previous": [],
694 "upcoming": [
695 {
696 "uid": "spotify:track:upcoming",
697 "source": "queue",
698 "item": {"uri": "spotify:track:upcoming", "entity_type": "track"},
699 }
700 ],
701 },
702 ):
703 ws.queue.put_nowait(_text_msg(payload))
704 ws.queue.put_nowait(None)
705 events: list[SoloistEvent] = []
706
707 async def on_event(event: SoloistEvent) -> None:
708 events.append(event)
709
710 await client.listen_events(on_event)
711
712 assert [event.type for event in events] == [
713 "auth_state",
714 "playback_state",
715 "volume_changed",
716 "position_sync",
717 "queue_changed",
718 ]
719 auth = events[0].data
720 assert isinstance(auth, SoloistAuthState)
721 assert auth.logged_in is True
722 assert auth.device_name == "Kitchen"
723 state = events[1].data
724 assert isinstance(state, SoloistPlaybackState)
725 assert state.status == "playing"
726 assert state.item is not None
727 assert state.item.uri == "spotify:track:2JRo0gjbX4GrCqBYdRohoo"
728 assert state.item.decorations["identity"]["name"] == "My Song"
729 assert state.options is not None
730 assert state.options.repeat == "off"
731 assert state.position is not None
732 assert state.position.position_ms == 45000
733 assert state.available_actions["seek_forward"]["step_ms"] == 15000
734 volume = events[2].data
735 assert isinstance(volume, SoloistVolumeChanged)
736 assert volume.volume == 42
737 sync = events[3].data
738 assert isinstance(sync, SoloistPositionSync)
739 assert sync.position.timestamp_ms == 1747654321000
740 queue = events[4].data
741 assert isinstance(queue, SoloistQueueChanged)
742 assert queue.upcoming[0].source == "queue"
743 assert queue.upcoming[0].item is not None
744 assert queue.upcoming[0].item.uri == "spotify:track:upcoming"
745
746
747async def test_the_item_boundary_events_decode_to_the_expected_payloads(tmp_path: Path) -> None:
748 """The events the playback backend cuts items on each decode to the payload it expects."""
749 _publish_endpoint(tmp_path)
750 ws = _FakeWebSocket()
751 client = _make_client(tmp_path, ws)
752 item = {"uri": "spotify:track:2JRo0gjbX4GrCqBYdRohoo", "entity_type": "track"}
753 for payload in (
754 {"type": "track_changed", "item": item},
755 {"type": "playback_state", "status": "playing", "item": item},
756 {"type": "playback_changed", "status": "playing", "item": item},
757 ):
758 ws.queue.put_nowait(_text_msg(payload))
759 ws.queue.put_nowait(None)
760 events: list[SoloistEvent] = []
761
762 async def on_event(event: SoloistEvent) -> None:
763 events.append(event)
764
765 await client.listen_events(on_event)
766
767 assert [event.type for event in events] == [
768 "track_changed",
769 "playback_state",
770 "playback_changed",
771 ]
772 assert isinstance(events[0].data, SoloistTrackChanged)
773 # the snapshot and the delta share one payload, so the event type is all a
774 # consumer has to tell a full state report from a partial one
775 assert isinstance(events[1].data, SoloistPlaybackState)
776 assert isinstance(events[2].data, SoloistPlaybackState)
777
778
779async def test_event_dispatch_tolerates_malformed_and_unknown(tmp_path: Path) -> None:
780 """Malformed frames are skipped, unknown event types pass through as raw events."""
781 _publish_endpoint(tmp_path)
782 ws = _FakeWebSocket()
783 client = _make_client(tmp_path, ws)
784 ws.queue.put_nowait(WSMessage(WSMsgType.TEXT, "not json", None))
785 ws.queue.put_nowait(WSMessage(WSMsgType.TEXT, '["a", "list"]', None))
786 ws.queue.put_nowait(_text_msg({"volume": 1})) # no type field
787 ws.queue.put_nowait(_text_msg({"type": "volume_changed"})) # missing required field
788 ws.queue.put_nowait(_text_msg({"type": "mystery_event", "foo": "bar"}))
789 ws.queue.put_nowait(_text_msg({"type": "volume_changed", "volume": 7}))
790 ws.queue.put_nowait(None)
791 events: list[SoloistEvent] = []
792
793 async def on_event(event: SoloistEvent) -> None:
794 events.append(event)
795
796 await client.listen_events(on_event)
797
798 assert len(events) == 2
799 assert events[0].type == "mystery_event"
800 assert events[0].data is None
801 assert events[0].raw == {"type": "mystery_event", "foo": "bar"}
802 volume = events[1].data
803 assert isinstance(volume, SoloistVolumeChanged)
804 assert volume.volume == 7
805
806
807async def _wait_connected(client: SoloistClient) -> None:
808 """Wait until the client's events WebSocket is connected."""
809 async with asyncio.timeout(5.0):
810 while not client.connected:
811 await asyncio.sleep(0.01)
812
813
814async def test_commands_have_documented_shape(tmp_path: Path) -> None:
815 """Command senders produce the documented wire frames (with clamped values)."""
816 _publish_endpoint(tmp_path)
817 ws = _FakeWebSocket()
818 client = _make_client(tmp_path, ws)
819
820 async def on_event(_event: SoloistEvent) -> None:
821 return
822
823 listen_task = asyncio.create_task(client.listen_events(on_event))
824 await _wait_connected(client)
825
826 await client.play("spotify:playlist:37i9dQZF1DXcBWIGoYBM5M")
827 await client.resume()
828 await client.pause()
829 await client.skip_next()
830 await client.seek(-100)
831 await client.set_volume(150)
832 await client.set_shuffle(True)
833 await client.set_repeat_context(True)
834 await client.add_to_queue("spotify:track:6rqhFgbbKwnb9MLmUQDhG6")
835 await client.get_queue(5)
836 ws.queue.put_nowait(None)
837 await listen_task
838
839 assert ws.sent == [
840 {"type": "command", "command": "play", "uri": "spotify:playlist:37i9dQZF1DXcBWIGoYBM5M"},
841 {"type": "command", "command": "play"},
842 {"type": "command", "command": "pause"},
843 {"type": "command", "command": "skip_next"},
844 {"type": "command", "command": "seek", "position_ms": 0},
845 {"type": "command", "command": "set_volume", "volume": 100},
846 {"type": "command", "command": "set_shuffle", "enabled": True},
847 {"type": "command", "command": "set_repeat_context", "enabled": True},
848 {
849 "type": "command",
850 "command": "add_to_queue",
851 "uri": "spotify:track:6rqhFgbbKwnb9MLmUQDhG6",
852 },
853 {"type": "command", "command": "get_queue", "limit": 5},
854 ]
855
856
857async def test_command_awaits_result(tmp_path: Path) -> None:
858 """A command sent with await_result resolves once its command_result arrives."""
859 _publish_endpoint(tmp_path)
860 ws = _FakeWebSocket()
861 client = _make_client(tmp_path, ws)
862 events: list[SoloistEvent] = []
863
864 async def on_event(event: SoloistEvent) -> None:
865 events.append(event)
866
867 listen_task = asyncio.create_task(client.listen_events(on_event))
868 await _wait_connected(client)
869
870 command_task = asyncio.create_task(client.activate(await_result=True))
871 await asyncio.sleep(0)
872 assert not command_task.done()
873 ws.queue.put_nowait(_text_msg({"type": "command_result", "command": "activate"}))
874 await asyncio.wait_for(command_task, timeout=1.0)
875 ws.queue.put_nowait(None)
876 await listen_task
877
878 # the ack is also still dispatched to the event callback
879 assert [event.type for event in events] == ["command_result"]
880
881
882async def test_commands_require_connection(tmp_path: Path) -> None:
883 """Sending a command without a connected WebSocket raises SoloistError."""
884 client = _make_client(tmp_path, _FakeWebSocket())
885
886 with pytest.raises(SoloistError, match="not connected"):
887 await client.pause()
888
889
890@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
891async def test_expired_by_timestamp_is_replaced(tmp_path: Path) -> None:
892 """An install past the 90-day build expiry is replaced even though it still runs."""
893 build_a = _build_archive(
894 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
895 )
896 build_b = _build_archive(
897 tmp_path / "b.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-B")}
898 )
899 manager, session = _make_manager(tmp_path, _serve_archive(build_a, etag='"v1"'))
900 await manager.ensure_binary(consent=True)
901 _age_metadata(tmp_path, days=100.0)
902 session.handler = _serve_archive(build_b, etag='"v2"')
903
904 path = await manager.ensure_fresh(consent=True)
905
906 assert b"GOOD-BUILD-B" in path.read_bytes()
907
908
909def test_parse_build_timestamp_reads_the_real_epoch_format() -> None:
910 """The observed 1.3.7 --version output carries the build date as a unix epoch."""
911 raw = "soloist 1.3.7.345 build 1787077868 (20260818) (gb24005ef46) (linux/aarch64)"
912 assert soloist._parse_build_timestamp(raw) == 1787077868.0
913 # an unrelated small number is not mistaken for a timestamp
914 assert soloist._parse_build_timestamp("soloist 1.2.3 build 42") is None
915
916
917@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
918async def test_force_refresh_bypasses_verification_cache(tmp_path: Path) -> None:
919 """force=True re-verifies even inside the recently-verified window (exit-10 path)."""
920 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
921 manager, session = _make_manager(tmp_path, _serve_archive(archive))
922 await manager.ensure_fresh(consent=True)
923 # age the install into the update window: a real re-verification is now
924 # observable as an update check against the CDN
925 _age_metadata(tmp_path)
926 # within the cache window a plain call still short-circuits...
927 session.requests.clear()
928 await manager.ensure_fresh(consent=True)
929 assert session.requests == []
930 # ...but a forced call re-verifies against the CDN (same build: no download)
931 await manager.ensure_fresh(consent=True, force=True)
932 assert soloist._last_verified is not None
933 assert [method for method, _ in session.requests] == ["HEAD"]
934