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