/
/
/
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 {"soloist": b"payload", "README": b"docs"}, # extra file
326 {"README": b"docs"}, # no soloist binary at all
327 ],
328)
329async def test_unsafe_or_unexpected_archive_rejected(
330 tmp_path: Path, files: dict[str, bytes]
331) -> None:
332 """Archives with traversal, absolute paths, extra or missing files are rejected."""
333 archive = _build_archive(tmp_path / "a.tar.gz", files)
334 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
335
336 with pytest.raises(InvalidArchiveError):
337 await manager.ensure_binary(consent=True)
338
339
340@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
341async def test_symlink_archive_rejected(tmp_path: Path) -> None:
342 """An archive delivering soloist as a symlink is rejected."""
343 archive = _build_archive(tmp_path / "a.tar.gz", symlink=("soloist", "/etc/passwd"))
344 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
345
346 with pytest.raises(InvalidArchiveError):
347 await manager.ensure_binary(consent=True)
348
349
350@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
351async def test_garbage_archive_rejected(tmp_path: Path) -> None:
352 """A response that is not a tar.gz archive at all is rejected."""
353 manager, _ = _make_manager(tmp_path, _serve_archive(b"this is not a tarball"))
354
355 with pytest.raises(InvalidArchiveError):
356 await manager.ensure_binary(consent=True)
357
358
359@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
360@pytest.mark.parametrize("content", [_elf_binary("arm64"), b"#!/bin/sh\necho not an elf\n"])
361async def test_wrong_or_non_elf_binary_rejected(tmp_path: Path, content: bytes) -> None:
362 """A binary for another architecture (or not an ELF at all) is rejected."""
363 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": content})
364 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
365
366 with pytest.raises(InvalidArchiveError):
367 await manager.ensure_binary(consent=True)
368 assert not manager.binary_path.exists()
369
370
371@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
372async def test_failed_validation_leaves_no_binary(tmp_path: Path) -> None:
373 """A fresh install whose binary fails --version validation leaves nothing behind."""
374 archive = _build_archive(
375 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"BROKEN")}
376 )
377 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
378
379 with pytest.raises(InvalidArchiveError):
380 await manager.ensure_binary(consent=True)
381 assert not manager.binary_path.exists()
382
383
384@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
385async def test_fresh_download_of_expired_build(tmp_path: Path) -> None:
386 """A freshly downloaded build that reports exit code 10 raises BuildExpiredError."""
387 archive = _build_archive(
388 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"EXPIRED")}
389 )
390 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
391
392 with pytest.raises(BuildExpiredError):
393 await manager.ensure_binary(consent=True)
394 assert not manager.binary_path.exists()
395
396
397@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
398async def test_rollback_restores_previous_binary(
399 tmp_path: Path, caplog: pytest.LogCaptureFixture
400) -> None:
401 """A failed replacement is rolled back to the previously installed binary."""
402 good = _build_archive(
403 tmp_path / "good.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
404 )
405 broken = _build_archive(
406 tmp_path / "broken.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"BROKEN")}
407 )
408 manager, session = _make_manager(tmp_path, _serve_archive(good, etag='"v1"'))
409 await manager.ensure_binary(consent=True)
410 _age_metadata(tmp_path)
411 session.handler = _serve_archive(broken, etag='"v2"')
412
413 with caplog.at_level(logging.WARNING):
414 path = await manager.ensure_fresh(consent=True)
415
416 assert path == manager.binary_path
417 assert b"GOOD-BUILD-A" in path.read_bytes()
418 assert not (_install_dir(tmp_path) / "soloist.prev").exists()
419 assert manager.diagnostics()["etag"] == '"v1"'
420 assert "keeping the current binary" in caplog.text
421
422
423@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
424async def test_refresh_installs_new_build(tmp_path: Path) -> None:
425 """An aged install is replaced when the CDN serves a different build."""
426 build_a = _build_archive(
427 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
428 )
429 build_b = _build_archive(
430 tmp_path / "b.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-B")}
431 )
432 manager, session = _make_manager(tmp_path, _serve_archive(build_a, etag='"v1"'))
433 await manager.ensure_binary(consent=True)
434 _age_metadata(tmp_path)
435 session.handler = _serve_archive(build_b, etag='"v2"')
436
437 path = await manager.ensure_fresh(consent=True)
438
439 assert b"GOOD-BUILD-B" in path.read_bytes()
440 diag = manager.diagnostics()
441 assert diag["etag"] == '"v2"'
442 assert diag["sha256"] == hashlib.sha256(build_b).hexdigest()
443
444
445@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
446async def test_refresh_skipped_when_etag_unchanged(tmp_path: Path) -> None:
447 """An aged install is kept when the CDN still serves the same build."""
448 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
449 manager, session = _make_manager(tmp_path, _serve_archive(archive, etag='"v1"'))
450 await manager.ensure_binary(consent=True)
451 _age_metadata(tmp_path)
452 session.requests.clear()
453
454 await manager.ensure_fresh(consent=True)
455
456 assert session.requests == [("HEAD", _CDN_URL.format(arch="x86_64"))]
457
458
459@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
460async def test_offline_refresh_returns_valid_binary(
461 tmp_path: Path, caplog: pytest.LogCaptureFixture
462) -> None:
463 """When offline, a still-valid installed binary is returned with a warning."""
464 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
465 manager, session = _make_manager(tmp_path, _serve_archive(archive))
466 await manager.ensure_binary(consent=True)
467 _age_metadata(tmp_path)
468 session.handler = _offline
469
470 with caplog.at_level(logging.WARNING):
471 path = await manager.ensure_fresh(consent=True)
472
473 assert path == manager.binary_path
474 assert "Unable to check for a soloist update" in caplog.text
475
476
477@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
478async def test_offline_with_expired_binary_raises(tmp_path: Path) -> None:
479 """When offline and the installed build already expired, BuildExpiredError is raised."""
480 install_dir = _install_dir(tmp_path)
481 install_dir.mkdir(parents=True)
482 (install_dir / "soloist").write_bytes(_elf_binary("x86_64", marker=b"EXPIRED"))
483 manager, _ = _make_manager(tmp_path, _offline)
484
485 with pytest.raises(BuildExpiredError):
486 await manager.ensure_fresh(consent=True)
487
488
489@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
490async def test_recent_verification_shared_across_managers(
491 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
492) -> None:
493 """Back-to-back ensure_fresh calls run the --version/CDN verification only once."""
494 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
495 manager1, session1 = _make_manager(tmp_path, _serve_archive(archive))
496 await manager1.ensure_binary(consent=True)
497 _age_metadata(tmp_path) # old enough that a full verification also HEADs the CDN
498 version_calls: list[str] = []
499
500 async def _counting_check_output(*args: str, **kwargs: Any) -> tuple[int, bytes]:
501 version_calls.append(args[0])
502 return await _fake_check_output(*args, **kwargs)
503
504 monkeypatch.setattr(soloist, "check_output", _counting_check_output)
505 session1.requests.clear()
506 manager2, session2 = _make_manager(tmp_path, _serve_archive(archive))
507
508 path1 = await manager1.ensure_fresh(consent=True)
509 path2 = await manager2.ensure_fresh(consent=True)
510
511 assert path1 == path2 == manager1.binary_path
512 # the second manager reuses the just-completed verification entirely
513 assert len(version_calls) == 1
514 assert session1.requests == [("HEAD", _CDN_URL.format(arch="x86_64"))]
515 assert session2.requests == []
516
517
518@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
519async def test_concurrent_callers_share_one_download(tmp_path: Path) -> None:
520 """Concurrent ensure_binary callers trigger exactly one download."""
521 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
522 manager, session = _make_manager(tmp_path, _serve_archive(archive))
523
524 paths = await asyncio.gather(*(manager.ensure_binary(consent=True) for _ in range(5)))
525
526 assert all(path == manager.binary_path for path in paths)
527 assert [req for req in session.requests if req[0] == "GET"] == [
528 ("GET", _CDN_URL.format(arch="x86_64"))
529 ]
530
531
532@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
533async def test_diagnostics_contains_no_secrets(tmp_path: Path) -> None:
534 """Diagnostics exposes install/build metadata only, never any key material."""
535 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
536 manager, _ = _make_manager(tmp_path, _serve_archive(archive, etag='"v1"'))
537 assert manager.diagnostics() == {"installed": False}
538
539 await manager.ensure_binary(consent=True)
540 diag = manager.diagnostics()
541
542 assert set(diag) == {
543 "installed",
544 "sha256",
545 "etag",
546 "version",
547 "version_raw",
548 "installed_at",
549 "build_timestamp",
550 "expires_at",
551 }
552 assert diag["installed"] is True
553 assert diag["sha256"] == hashlib.sha256(archive).hexdigest()
554 assert diag["etag"] == '"v1"'
555 assert diag["version"] == "1.2.3"
556 assert diag["expires_at"] == pytest.approx(diag["build_timestamp"] + 90 * 24 * 3600)
557
558
559class _FakeWebSocket:
560 """Fake events WebSocket: an async iterator fed from a queue, recording sent frames."""
561
562 def __init__(self) -> None:
563 self.queue: asyncio.Queue[WSMessage | None] = asyncio.Queue()
564 self.sent: list[dict[str, Any]] = []
565 self.closed = False
566
567 async def __aenter__(self) -> Self:
568 return self
569
570 async def __aexit__(self, *exc_info: object) -> None:
571 self.closed = True
572
573 def __aiter__(self) -> _FakeWebSocket:
574 return self
575
576 async def __anext__(self) -> WSMessage:
577 msg = await self.queue.get()
578 if msg is None:
579 raise StopAsyncIteration
580 return msg
581
582 async def send_json(self, data: dict[str, Any]) -> None:
583 """Record an outgoing JSON frame."""
584 self.sent.append(data)
585
586 def exception(self) -> BaseException | None:
587 """Return the connection error (never set for this fake)."""
588 return None
589
590
591def _make_client(data_dir: Path, ws: _FakeWebSocket) -> SoloistClient:
592 """Create a client for the given data dir whose session connects to the fake ws."""
593 mass = SimpleNamespace(http_session=SimpleNamespace(ws_connect=lambda *_a, **_kw: ws))
594 return SoloistClient(cast("MusicAssistant", mass), data_dir, logging.getLogger("test.soloist"))
595
596
597def _publish_endpoint(data_dir: Path, addr: str = "127.0.0.1", port: str = "8765") -> None:
598 """Write the ws.addr/ws.port endpoint files like the daemon does."""
599 (data_dir / "ws.addr").write_text(f"{addr}\n", encoding="utf-8")
600 (data_dir / "ws.port").write_text(f"{port}\n", encoding="utf-8")
601
602
603def _text_msg(payload: dict[str, Any]) -> WSMessage:
604 """Wrap an event payload in a WebSocket TEXT message."""
605 return WSMessage(WSMsgType.TEXT, json.dumps(payload), None)
606
607
608async def test_endpoint_discovery_polls_until_ready(
609 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
610) -> None:
611 """Endpoint discovery keeps polling until both files exist and parse."""
612 monkeypatch.setattr(soloist, "_ENDPOINT_POLL_INTERVAL", 0.01)
613 client = _make_client(tmp_path, _FakeWebSocket())
614 task = asyncio.create_task(client.wait_until_ready(timeout=5.0))
615
616 await asyncio.sleep(0.05)
617 assert not task.done()
618 (tmp_path / "ws.addr").write_text("127.0.0.1\n", encoding="utf-8")
619 (tmp_path / "ws.port").write_text("not-a-port\n", encoding="utf-8")
620 await asyncio.sleep(0.05)
621 assert not task.done() # unparsable port: keep polling
622 (tmp_path / "ws.port").write_text("8765\n", encoding="utf-8")
623
624 assert await task is True
625
626
627async def test_endpoint_discovery_times_out(
628 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
629) -> None:
630 """Endpoint discovery returns False when the files never appear."""
631 monkeypatch.setattr(soloist, "_ENDPOINT_POLL_INTERVAL", 0.01)
632 client = _make_client(tmp_path, _FakeWebSocket())
633
634 assert await client.wait_until_ready(timeout=0.05) is False
635
636
637async def test_event_dispatch_decodes_documented_payloads(tmp_path: Path) -> None:
638 """Documented event payloads are decoded into their typed models."""
639 _publish_endpoint(tmp_path)
640 ws = _FakeWebSocket()
641 client = _make_client(tmp_path, ws)
642 playback_state = {
643 "type": "playback_state",
644 "status": "playing",
645 "item": {
646 "uri": "spotify:track:2JRo0gjbX4GrCqBYdRohoo",
647 "entity_type": "track",
648 "decorations": {
649 "identity": {"name": "My Song"},
650 "playback": {"duration_ms": 210000, "content_ratings": []},
651 },
652 },
653 "context": {
654 "uri": "spotify:playlist:37i9dQZF1DXcBWIGoYBM5M",
655 "entity_type": "playlist",
656 "decorations": {"identity": {"name": "Today's Top Hits"}},
657 },
658 "position": {"position_ms": 45000, "timestamp_ms": 1747654321000, "speed": 1.0},
659 "volume": 65,
660 "is_active": True,
661 "options": {"shuffle": False, "repeat": "off", "playback_speed": 1.0, "modes": {}},
662 "available_actions": {"pause": {}, "seek_forward": {"step_ms": 15000}},
663 }
664 for payload in (
665 {"type": "auth_state", "logged_in": True, "is_active": True, "device_name": "Kitchen"},
666 playback_state,
667 {"type": "volume_changed", "volume": 42},
668 {
669 "type": "position_sync",
670 "position": {"position_ms": 45000, "timestamp_ms": 1747654321000, "speed": 1.0},
671 },
672 {
673 "type": "queue_changed",
674 "previous": [],
675 "upcoming": [
676 {
677 "uid": "spotify:track:upcoming",
678 "source": "queue",
679 "item": {"uri": "spotify:track:upcoming", "entity_type": "track"},
680 }
681 ],
682 },
683 ):
684 ws.queue.put_nowait(_text_msg(payload))
685 ws.queue.put_nowait(None)
686 events: list[SoloistEvent] = []
687
688 async def on_event(event: SoloistEvent) -> None:
689 events.append(event)
690
691 await client.listen_events(on_event)
692
693 assert [event.type for event in events] == [
694 "auth_state",
695 "playback_state",
696 "volume_changed",
697 "position_sync",
698 "queue_changed",
699 ]
700 auth = events[0].data
701 assert isinstance(auth, SoloistAuthState)
702 assert auth.logged_in is True
703 assert auth.device_name == "Kitchen"
704 state = events[1].data
705 assert isinstance(state, SoloistPlaybackState)
706 assert state.status == "playing"
707 assert state.item is not None
708 assert state.item.uri == "spotify:track:2JRo0gjbX4GrCqBYdRohoo"
709 assert state.item.decorations["identity"]["name"] == "My Song"
710 assert state.options is not None
711 assert state.options.repeat == "off"
712 assert state.position is not None
713 assert state.position.position_ms == 45000
714 assert state.available_actions["seek_forward"]["step_ms"] == 15000
715 volume = events[2].data
716 assert isinstance(volume, SoloistVolumeChanged)
717 assert volume.volume == 42
718 sync = events[3].data
719 assert isinstance(sync, SoloistPositionSync)
720 assert sync.position.timestamp_ms == 1747654321000
721 queue = events[4].data
722 assert isinstance(queue, SoloistQueueChanged)
723 assert queue.upcoming[0].source == "queue"
724 assert queue.upcoming[0].item is not None
725 assert queue.upcoming[0].item.uri == "spotify:track:upcoming"
726
727
728async def test_the_item_boundary_events_decode_to_the_expected_payloads(tmp_path: Path) -> None:
729 """The events the playback backend cuts items on each decode to the payload it expects."""
730 _publish_endpoint(tmp_path)
731 ws = _FakeWebSocket()
732 client = _make_client(tmp_path, ws)
733 item = {"uri": "spotify:track:2JRo0gjbX4GrCqBYdRohoo", "entity_type": "track"}
734 for payload in (
735 {"type": "track_changed", "item": item},
736 {"type": "playback_state", "status": "playing", "item": item},
737 {"type": "playback_changed", "status": "playing", "item": item},
738 ):
739 ws.queue.put_nowait(_text_msg(payload))
740 ws.queue.put_nowait(None)
741 events: list[SoloistEvent] = []
742
743 async def on_event(event: SoloistEvent) -> None:
744 events.append(event)
745
746 await client.listen_events(on_event)
747
748 assert [event.type for event in events] == [
749 "track_changed",
750 "playback_state",
751 "playback_changed",
752 ]
753 assert isinstance(events[0].data, SoloistTrackChanged)
754 # the snapshot and the delta share one payload, so the event type is all a
755 # consumer has to tell a full state report from a partial one
756 assert isinstance(events[1].data, SoloistPlaybackState)
757 assert isinstance(events[2].data, SoloistPlaybackState)
758
759
760async def test_event_dispatch_tolerates_malformed_and_unknown(tmp_path: Path) -> None:
761 """Malformed frames are skipped, unknown event types pass through as raw events."""
762 _publish_endpoint(tmp_path)
763 ws = _FakeWebSocket()
764 client = _make_client(tmp_path, ws)
765 ws.queue.put_nowait(WSMessage(WSMsgType.TEXT, "not json", None))
766 ws.queue.put_nowait(WSMessage(WSMsgType.TEXT, '["a", "list"]', None))
767 ws.queue.put_nowait(_text_msg({"volume": 1})) # no type field
768 ws.queue.put_nowait(_text_msg({"type": "volume_changed"})) # missing required field
769 ws.queue.put_nowait(_text_msg({"type": "mystery_event", "foo": "bar"}))
770 ws.queue.put_nowait(_text_msg({"type": "volume_changed", "volume": 7}))
771 ws.queue.put_nowait(None)
772 events: list[SoloistEvent] = []
773
774 async def on_event(event: SoloistEvent) -> None:
775 events.append(event)
776
777 await client.listen_events(on_event)
778
779 assert len(events) == 2
780 assert events[0].type == "mystery_event"
781 assert events[0].data is None
782 assert events[0].raw == {"type": "mystery_event", "foo": "bar"}
783 volume = events[1].data
784 assert isinstance(volume, SoloistVolumeChanged)
785 assert volume.volume == 7
786
787
788async def _wait_connected(client: SoloistClient) -> None:
789 """Wait until the client's events WebSocket is connected."""
790 async with asyncio.timeout(5.0):
791 while not client.connected:
792 await asyncio.sleep(0.01)
793
794
795async def test_commands_have_documented_shape(tmp_path: Path) -> None:
796 """Command senders produce the documented wire frames (with clamped values)."""
797 _publish_endpoint(tmp_path)
798 ws = _FakeWebSocket()
799 client = _make_client(tmp_path, ws)
800
801 async def on_event(_event: SoloistEvent) -> None:
802 return
803
804 listen_task = asyncio.create_task(client.listen_events(on_event))
805 await _wait_connected(client)
806
807 await client.play("spotify:playlist:37i9dQZF1DXcBWIGoYBM5M")
808 await client.resume()
809 await client.pause()
810 await client.skip_next()
811 await client.seek(-100)
812 await client.set_volume(150)
813 await client.set_shuffle(True)
814 await client.set_repeat_context(True)
815 await client.add_to_queue("spotify:track:6rqhFgbbKwnb9MLmUQDhG6")
816 await client.get_queue(5)
817 ws.queue.put_nowait(None)
818 await listen_task
819
820 assert ws.sent == [
821 {"type": "command", "command": "play", "uri": "spotify:playlist:37i9dQZF1DXcBWIGoYBM5M"},
822 {"type": "command", "command": "play"},
823 {"type": "command", "command": "pause"},
824 {"type": "command", "command": "skip_next"},
825 {"type": "command", "command": "seek", "position_ms": 0},
826 {"type": "command", "command": "set_volume", "volume": 100},
827 {"type": "command", "command": "set_shuffle", "enabled": True},
828 {"type": "command", "command": "set_repeat_context", "enabled": True},
829 {
830 "type": "command",
831 "command": "add_to_queue",
832 "uri": "spotify:track:6rqhFgbbKwnb9MLmUQDhG6",
833 },
834 {"type": "command", "command": "get_queue", "limit": 5},
835 ]
836
837
838async def test_command_awaits_result(tmp_path: Path) -> None:
839 """A command sent with await_result resolves once its command_result arrives."""
840 _publish_endpoint(tmp_path)
841 ws = _FakeWebSocket()
842 client = _make_client(tmp_path, ws)
843 events: list[SoloistEvent] = []
844
845 async def on_event(event: SoloistEvent) -> None:
846 events.append(event)
847
848 listen_task = asyncio.create_task(client.listen_events(on_event))
849 await _wait_connected(client)
850
851 command_task = asyncio.create_task(client.activate(await_result=True))
852 await asyncio.sleep(0)
853 assert not command_task.done()
854 ws.queue.put_nowait(_text_msg({"type": "command_result", "command": "activate"}))
855 await asyncio.wait_for(command_task, timeout=1.0)
856 ws.queue.put_nowait(None)
857 await listen_task
858
859 # the ack is also still dispatched to the event callback
860 assert [event.type for event in events] == ["command_result"]
861
862
863async def test_commands_require_connection(tmp_path: Path) -> None:
864 """Sending a command without a connected WebSocket raises SoloistError."""
865 client = _make_client(tmp_path, _FakeWebSocket())
866
867 with pytest.raises(SoloistError, match="not connected"):
868 await client.pause()
869
870
871@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
872async def test_expired_by_timestamp_is_replaced(tmp_path: Path) -> None:
873 """An install past the 90-day build expiry is replaced even though it still runs."""
874 build_a = _build_archive(
875 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
876 )
877 build_b = _build_archive(
878 tmp_path / "b.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-B")}
879 )
880 manager, session = _make_manager(tmp_path, _serve_archive(build_a, etag='"v1"'))
881 await manager.ensure_binary(consent=True)
882 _age_metadata(tmp_path, days=100.0)
883 session.handler = _serve_archive(build_b, etag='"v2"')
884
885 path = await manager.ensure_fresh(consent=True)
886
887 assert b"GOOD-BUILD-B" in path.read_bytes()
888
889
890def test_parse_build_timestamp_reads_the_real_epoch_format() -> None:
891 """The observed 1.3.7 --version output carries the build date as a unix epoch."""
892 raw = "soloist 1.3.7.345 build 1787077868 (20260818) (gb24005ef46) (linux/aarch64)"
893 assert soloist._parse_build_timestamp(raw) == 1787077868.0
894 # an unrelated small number is not mistaken for a timestamp
895 assert soloist._parse_build_timestamp("soloist 1.2.3 build 42") is None
896
897
898@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
899async def test_force_refresh_bypasses_verification_cache(tmp_path: Path) -> None:
900 """force=True re-verifies even inside the recently-verified window (exit-10 path)."""
901 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
902 manager, session = _make_manager(tmp_path, _serve_archive(archive))
903 await manager.ensure_fresh(consent=True)
904 # age the install into the update window: a real re-verification is now
905 # observable as an update check against the CDN
906 _age_metadata(tmp_path)
907 # within the cache window a plain call still short-circuits...
908 session.requests.clear()
909 await manager.ensure_fresh(consent=True)
910 assert session.requests == []
911 # ...but a forced call re-verifies against the CDN (same build: no download)
912 await manager.ensure_fresh(consent=True, force=True)
913 assert soloist._last_verified is not None
914 assert [method for method, _ in session.requests] == ["HEAD"]
915