/
/
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 import soloist
23from music_assistant.providers.spotify_connect.soloist 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
196def linux_platform(monkeypatch: pytest.MonkeyPatch) -> None:
197 """Pretend to run on Linux x86_64 (tests run on macOS)."""
198 monkeypatch.setattr(platform, "system", lambda: "Linux")
199 monkeypatch.setattr(platform, "machine", lambda: "x86_64")
200
201
202@pytest.mark.usefixtures("fake_version_cmd")
203@pytest.mark.parametrize(
204 ("machine", "arch"),
205 [
206 ("aarch64", "arm64"),
207 ("arm64", "arm64"),
208 ("armv7l", "arm32"),
209 ("armv8l", "arm32"),
210 ("x86_64", "x86_64"),
211 ("amd64", "x86_64"),
212 ],
213)
214async def test_arch_maps_to_cdn_artifact(
215 machine: str, arch: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
216) -> None:
217 """Each supported machine downloads the matching CDN artifact."""
218 monkeypatch.setattr(platform, "system", lambda: "Linux")
219 monkeypatch.setattr(platform, "machine", lambda: machine)
220 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary(arch)})
221 manager, session = _make_manager(tmp_path, _serve_archive(archive))
222
223 path = await manager.ensure_binary(consent=True)
224
225 assert path.is_file()
226 assert session.requests == [("GET", _CDN_URL.format(arch=arch))]
227
228
229async def test_non_linux_platform_rejected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
230 """A non-Linux platform is rejected without any network access."""
231 monkeypatch.setattr(platform, "system", lambda: "Darwin")
232 monkeypatch.setattr(platform, "machine", lambda: "arm64")
233 manager, session = _make_manager(tmp_path, _offline)
234
235 with pytest.raises(UnsupportedPlatformError):
236 await manager.ensure_binary(consent=True)
237 assert session.requests == []
238
239
240async def test_unknown_machine_rejected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
241 """An unknown machine architecture is rejected without any network access."""
242 monkeypatch.setattr(platform, "system", lambda: "Linux")
243 monkeypatch.setattr(platform, "machine", lambda: "mips64")
244 manager, session = _make_manager(tmp_path, _offline)
245
246 with pytest.raises(UnsupportedPlatformError):
247 await manager.ensure_binary(consent=True)
248 assert session.requests == []
249
250
251@pytest.mark.usefixtures("linux_platform")
252async def test_download_requires_consent(tmp_path: Path) -> None:
253 """Without consent no download is attempted and no network call is made."""
254 manager, session = _make_manager(tmp_path, _offline)
255
256 with pytest.raises(ConsentRequiredError):
257 await manager.ensure_binary(consent=False)
258 assert session.requests == []
259
260
261@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
262async def test_installed_binary_returned_without_network(tmp_path: Path) -> None:
263 """An already-installed valid binary is returned without consent or network."""
264 install_dir = _install_dir(tmp_path)
265 install_dir.mkdir(parents=True)
266 (install_dir / "soloist").write_bytes(_elf_binary("x86_64"))
267 manager, session = _make_manager(tmp_path, _offline)
268
269 path = await manager.ensure_binary(consent=False)
270
271 assert path == install_dir / "soloist"
272 assert session.requests == []
273
274
275@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
276@pytest.mark.parametrize("redirect_host", ["evil.example.com", "evilspotifycdn.com"])
277async def test_redirect_outside_allowlist_rejected(tmp_path: Path, redirect_host: str) -> None:
278 """A redirect to a host outside Spotify's infrastructure aborts the download."""
279
280 def handler(_method: str, _url: str) -> _FakeResponse:
281 return _FakeResponse(
282 status=302, headers={"Location": f"https://{redirect_host}/soloist.tar.gz"}
283 )
284
285 manager, session = _make_manager(tmp_path, handler)
286
287 with pytest.raises(DownloadFailedError, match="untrusted host"):
288 await manager.ensure_binary(consent=True)
289 # only the initial request went out, the redirect was never followed
290 assert len(session.requests) == 1
291
292
293@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
294async def test_redirect_within_allowlist_followed(tmp_path: Path) -> None:
295 """A redirect within Spotify's infrastructure is followed and the download succeeds."""
296 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
297 redirect_url = "https://downloads.spotify.com/soloist_release_x86_64.tar.gz"
298
299 def handler(_method: str, url: str) -> _FakeResponse:
300 if url != redirect_url:
301 return _FakeResponse(status=302, headers={"Location": redirect_url})
302 return _FakeResponse(body=archive)
303
304 manager, session = _make_manager(tmp_path, handler)
305
306 path = await manager.ensure_binary(consent=True)
307
308 assert path.is_file()
309 assert session.requests[-1] == ("GET", redirect_url)
310
311
312@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
313@pytest.mark.parametrize(
314 "files",
315 [
316 {"../soloist": b"payload"}, # path traversal
317 {"/soloist": b"payload"}, # absolute path
318 {"soloist": b"payload", "README": b"docs"}, # extra file
319 {"README": b"docs"}, # no soloist binary at all
320 ],
321)
322async def test_unsafe_or_unexpected_archive_rejected(
323 tmp_path: Path, files: dict[str, bytes]
324) -> None:
325 """Archives with traversal, absolute paths, extra or missing files are rejected."""
326 archive = _build_archive(tmp_path / "a.tar.gz", files)
327 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
328
329 with pytest.raises(InvalidArchiveError):
330 await manager.ensure_binary(consent=True)
331
332
333@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
334async def test_symlink_archive_rejected(tmp_path: Path) -> None:
335 """An archive delivering soloist as a symlink is rejected."""
336 archive = _build_archive(tmp_path / "a.tar.gz", symlink=("soloist", "/etc/passwd"))
337 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
338
339 with pytest.raises(InvalidArchiveError):
340 await manager.ensure_binary(consent=True)
341
342
343@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
344async def test_garbage_archive_rejected(tmp_path: Path) -> None:
345 """A response that is not a tar.gz archive at all is rejected."""
346 manager, _ = _make_manager(tmp_path, _serve_archive(b"this is not a tarball"))
347
348 with pytest.raises(InvalidArchiveError):
349 await manager.ensure_binary(consent=True)
350
351
352@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
353@pytest.mark.parametrize("content", [_elf_binary("arm64"), b"#!/bin/sh\necho not an elf\n"])
354async def test_wrong_or_non_elf_binary_rejected(tmp_path: Path, content: bytes) -> None:
355 """A binary for another architecture (or not an ELF at all) is rejected."""
356 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": content})
357 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
358
359 with pytest.raises(InvalidArchiveError):
360 await manager.ensure_binary(consent=True)
361 assert not manager.binary_path.exists()
362
363
364@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
365async def test_failed_validation_leaves_no_binary(tmp_path: Path) -> None:
366 """A fresh install whose binary fails --version validation leaves nothing behind."""
367 archive = _build_archive(
368 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"BROKEN")}
369 )
370 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
371
372 with pytest.raises(InvalidArchiveError):
373 await manager.ensure_binary(consent=True)
374 assert not manager.binary_path.exists()
375
376
377@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
378async def test_fresh_download_of_expired_build(tmp_path: Path) -> None:
379 """A freshly downloaded build that reports exit code 10 raises BuildExpiredError."""
380 archive = _build_archive(
381 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"EXPIRED")}
382 )
383 manager, _ = _make_manager(tmp_path, _serve_archive(archive))
384
385 with pytest.raises(BuildExpiredError):
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_rollback_restores_previous_binary(
392 tmp_path: Path, caplog: pytest.LogCaptureFixture
393) -> None:
394 """A failed replacement is rolled back to the previously installed binary."""
395 good = _build_archive(
396 tmp_path / "good.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
397 )
398 broken = _build_archive(
399 tmp_path / "broken.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"BROKEN")}
400 )
401 manager, session = _make_manager(tmp_path, _serve_archive(good, etag='"v1"'))
402 await manager.ensure_binary(consent=True)
403 _age_metadata(tmp_path)
404 session.handler = _serve_archive(broken, etag='"v2"')
405
406 with caplog.at_level(logging.WARNING):
407 path = await manager.ensure_fresh(consent=True)
408
409 assert path == manager.binary_path
410 assert b"GOOD-BUILD-A" in path.read_bytes()
411 assert not (_install_dir(tmp_path) / "soloist.prev").exists()
412 assert manager.diagnostics()["etag"] == '"v1"'
413 assert "keeping the current binary" in caplog.text
414
415
416@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
417async def test_refresh_installs_new_build(tmp_path: Path) -> None:
418 """An aged install is replaced when the CDN serves a different build."""
419 build_a = _build_archive(
420 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
421 )
422 build_b = _build_archive(
423 tmp_path / "b.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-B")}
424 )
425 manager, session = _make_manager(tmp_path, _serve_archive(build_a, etag='"v1"'))
426 await manager.ensure_binary(consent=True)
427 _age_metadata(tmp_path)
428 session.handler = _serve_archive(build_b, etag='"v2"')
429
430 path = await manager.ensure_fresh(consent=True)
431
432 assert b"GOOD-BUILD-B" in path.read_bytes()
433 diag = manager.diagnostics()
434 assert diag["etag"] == '"v2"'
435 assert diag["sha256"] == hashlib.sha256(build_b).hexdigest()
436
437
438@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
439async def test_refresh_skipped_when_etag_unchanged(tmp_path: Path) -> None:
440 """An aged install is kept when the CDN still serves the same build."""
441 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
442 manager, session = _make_manager(tmp_path, _serve_archive(archive, etag='"v1"'))
443 await manager.ensure_binary(consent=True)
444 _age_metadata(tmp_path)
445 session.requests.clear()
446
447 await manager.ensure_fresh(consent=True)
448
449 assert session.requests == [("HEAD", _CDN_URL.format(arch="x86_64"))]
450
451
452@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
453async def test_offline_refresh_returns_valid_binary(
454 tmp_path: Path, caplog: pytest.LogCaptureFixture
455) -> None:
456 """When offline, a still-valid installed binary is returned with a warning."""
457 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
458 manager, session = _make_manager(tmp_path, _serve_archive(archive))
459 await manager.ensure_binary(consent=True)
460 _age_metadata(tmp_path)
461 session.handler = _offline
462
463 with caplog.at_level(logging.WARNING):
464 path = await manager.ensure_fresh(consent=True)
465
466 assert path == manager.binary_path
467 assert "Unable to check for a soloist update" in caplog.text
468
469
470@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
471async def test_offline_with_expired_binary_raises(tmp_path: Path) -> None:
472 """When offline and the installed build already expired, BuildExpiredError is raised."""
473 install_dir = _install_dir(tmp_path)
474 install_dir.mkdir(parents=True)
475 (install_dir / "soloist").write_bytes(_elf_binary("x86_64", marker=b"EXPIRED"))
476 manager, _ = _make_manager(tmp_path, _offline)
477
478 with pytest.raises(BuildExpiredError):
479 await manager.ensure_fresh(consent=True)
480
481
482@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
483async def test_concurrent_callers_share_one_download(tmp_path: Path) -> None:
484 """Concurrent ensure_binary callers trigger exactly one download."""
485 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
486 manager, session = _make_manager(tmp_path, _serve_archive(archive))
487
488 paths = await asyncio.gather(*(manager.ensure_binary(consent=True) for _ in range(5)))
489
490 assert all(path == manager.binary_path for path in paths)
491 assert [req for req in session.requests if req[0] == "GET"] == [
492 ("GET", _CDN_URL.format(arch="x86_64"))
493 ]
494
495
496@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
497async def test_diagnostics_contains_no_secrets(tmp_path: Path) -> None:
498 """Diagnostics exposes install/build metadata only, never any key material."""
499 archive = _build_archive(tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64")})
500 manager, _ = _make_manager(tmp_path, _serve_archive(archive, etag='"v1"'))
501 assert manager.diagnostics() == {"installed": False}
502
503 await manager.ensure_binary(consent=True)
504 diag = manager.diagnostics()
505
506 assert set(diag) == {
507 "installed",
508 "sha256",
509 "etag",
510 "version",
511 "version_raw",
512 "installed_at",
513 "build_timestamp",
514 "expires_at",
515 }
516 assert diag["installed"] is True
517 assert diag["sha256"] == hashlib.sha256(archive).hexdigest()
518 assert diag["etag"] == '"v1"'
519 assert diag["version"] == "1.2.3"
520 assert diag["expires_at"] == pytest.approx(diag["build_timestamp"] + 90 * 24 * 3600)
521
522
523class _FakeWebSocket:
524 """Fake events WebSocket: an async iterator fed from a queue, recording sent frames."""
525
526 def __init__(self) -> None:
527 self.queue: asyncio.Queue[WSMessage | None] = asyncio.Queue()
528 self.sent: list[dict[str, Any]] = []
529 self.closed = False
530
531 async def __aenter__(self) -> Self:
532 return self
533
534 async def __aexit__(self, *exc_info: object) -> None:
535 self.closed = True
536
537 def __aiter__(self) -> _FakeWebSocket:
538 return self
539
540 async def __anext__(self) -> WSMessage:
541 msg = await self.queue.get()
542 if msg is None:
543 raise StopAsyncIteration
544 return msg
545
546 async def send_json(self, data: dict[str, Any]) -> None:
547 """Record an outgoing JSON frame."""
548 self.sent.append(data)
549
550 def exception(self) -> BaseException | None:
551 """Return the connection error (never set for this fake)."""
552 return None
553
554
555def _make_client(data_dir: Path, ws: _FakeWebSocket) -> SoloistClient:
556 """Create a client for the given data dir whose session connects to the fake ws."""
557 mass = SimpleNamespace(http_session=SimpleNamespace(ws_connect=lambda *_a, **_kw: ws))
558 return SoloistClient(cast("MusicAssistant", mass), data_dir, logging.getLogger("test.soloist"))
559
560
561def _publish_endpoint(data_dir: Path, addr: str = "127.0.0.1", port: str = "8765") -> None:
562 """Write the ws.addr/ws.port endpoint files like the daemon does."""
563 (data_dir / "ws.addr").write_text(f"{addr}\n", encoding="utf-8")
564 (data_dir / "ws.port").write_text(f"{port}\n", encoding="utf-8")
565
566
567def _text_msg(payload: dict[str, Any]) -> WSMessage:
568 """Wrap an event payload in a WebSocket TEXT message."""
569 return WSMessage(WSMsgType.TEXT, json.dumps(payload), None)
570
571
572async def test_endpoint_discovery_polls_until_ready(
573 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
574) -> None:
575 """Endpoint discovery keeps polling until both files exist and parse."""
576 monkeypatch.setattr(soloist, "_ENDPOINT_POLL_INTERVAL", 0.01)
577 client = _make_client(tmp_path, _FakeWebSocket())
578 task = asyncio.create_task(client.wait_until_ready(timeout=5.0))
579
580 await asyncio.sleep(0.05)
581 assert not task.done()
582 (tmp_path / "ws.addr").write_text("127.0.0.1\n", encoding="utf-8")
583 (tmp_path / "ws.port").write_text("not-a-port\n", encoding="utf-8")
584 await asyncio.sleep(0.05)
585 assert not task.done() # unparsable port: keep polling
586 (tmp_path / "ws.port").write_text("8765\n", encoding="utf-8")
587
588 assert await task is True
589
590
591async def test_endpoint_discovery_times_out(
592 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
593) -> None:
594 """Endpoint discovery returns False when the files never appear."""
595 monkeypatch.setattr(soloist, "_ENDPOINT_POLL_INTERVAL", 0.01)
596 client = _make_client(tmp_path, _FakeWebSocket())
597
598 assert await client.wait_until_ready(timeout=0.05) is False
599
600
601async def test_event_dispatch_decodes_documented_payloads(tmp_path: Path) -> None:
602 """Documented event payloads are decoded into their typed models."""
603 _publish_endpoint(tmp_path)
604 ws = _FakeWebSocket()
605 client = _make_client(tmp_path, ws)
606 playback_state = {
607 "type": "playback_state",
608 "status": "playing",
609 "item": {
610 "uri": "spotify:track:2JRo0gjbX4GrCqBYdRohoo",
611 "entity_type": "track",
612 "decorations": {
613 "identity": {"name": "My Song"},
614 "playback": {"duration_ms": 210000, "content_ratings": []},
615 },
616 },
617 "context": {
618 "uri": "spotify:playlist:37i9dQZF1DXcBWIGoYBM5M",
619 "entity_type": "playlist",
620 "decorations": {"identity": {"name": "Today's Top Hits"}},
621 },
622 "position": {"position_ms": 45000, "timestamp_ms": 1747654321000, "speed": 1.0},
623 "volume": 65,
624 "is_active": True,
625 "options": {"shuffle": False, "repeat": "off", "playback_speed": 1.0, "modes": {}},
626 "available_actions": {"pause": {}, "seek_forward": {"step_ms": 15000}},
627 }
628 for payload in (
629 {"type": "auth_state", "logged_in": True, "is_active": True, "device_name": "Kitchen"},
630 playback_state,
631 {"type": "volume_changed", "volume": 42},
632 {
633 "type": "position_sync",
634 "position": {"position_ms": 45000, "timestamp_ms": 1747654321000, "speed": 1.0},
635 },
636 {
637 "type": "queue_changed",
638 "previous": [],
639 "upcoming": [
640 {
641 "uid": "spotify:track:upcoming",
642 "source": "queue",
643 "item": {"uri": "spotify:track:upcoming", "entity_type": "track"},
644 }
645 ],
646 },
647 ):
648 ws.queue.put_nowait(_text_msg(payload))
649 ws.queue.put_nowait(None)
650 events: list[SoloistEvent] = []
651
652 async def on_event(event: SoloistEvent) -> None:
653 events.append(event)
654
655 await client.listen_events(on_event)
656
657 assert [event.type for event in events] == [
658 "auth_state",
659 "playback_state",
660 "volume_changed",
661 "position_sync",
662 "queue_changed",
663 ]
664 auth = events[0].data
665 assert isinstance(auth, SoloistAuthState)
666 assert auth.logged_in is True
667 assert auth.device_name == "Kitchen"
668 state = events[1].data
669 assert isinstance(state, SoloistPlaybackState)
670 assert state.status == "playing"
671 assert state.item is not None
672 assert state.item.uri == "spotify:track:2JRo0gjbX4GrCqBYdRohoo"
673 assert state.item.decorations["identity"]["name"] == "My Song"
674 assert state.options is not None
675 assert state.options.repeat == "off"
676 assert state.position is not None
677 assert state.position.position_ms == 45000
678 assert state.available_actions["seek_forward"]["step_ms"] == 15000
679 volume = events[2].data
680 assert isinstance(volume, SoloistVolumeChanged)
681 assert volume.volume == 42
682 sync = events[3].data
683 assert isinstance(sync, SoloistPositionSync)
684 assert sync.position.timestamp_ms == 1747654321000
685 queue = events[4].data
686 assert isinstance(queue, SoloistQueueChanged)
687 assert queue.upcoming[0].source == "queue"
688 assert queue.upcoming[0].item is not None
689 assert queue.upcoming[0].item.uri == "spotify:track:upcoming"
690
691
692async def test_event_dispatch_tolerates_malformed_and_unknown(tmp_path: Path) -> None:
693 """Malformed frames are skipped, unknown event types pass through as raw events."""
694 _publish_endpoint(tmp_path)
695 ws = _FakeWebSocket()
696 client = _make_client(tmp_path, ws)
697 ws.queue.put_nowait(WSMessage(WSMsgType.TEXT, "not json", None))
698 ws.queue.put_nowait(WSMessage(WSMsgType.TEXT, '["a", "list"]', None))
699 ws.queue.put_nowait(_text_msg({"volume": 1})) # no type field
700 ws.queue.put_nowait(_text_msg({"type": "volume_changed"})) # missing required field
701 ws.queue.put_nowait(_text_msg({"type": "mystery_event", "foo": "bar"}))
702 ws.queue.put_nowait(_text_msg({"type": "volume_changed", "volume": 7}))
703 ws.queue.put_nowait(None)
704 events: list[SoloistEvent] = []
705
706 async def on_event(event: SoloistEvent) -> None:
707 events.append(event)
708
709 await client.listen_events(on_event)
710
711 assert len(events) == 2
712 assert events[0].type == "mystery_event"
713 assert events[0].data is None
714 assert events[0].raw == {"type": "mystery_event", "foo": "bar"}
715 volume = events[1].data
716 assert isinstance(volume, SoloistVolumeChanged)
717 assert volume.volume == 7
718
719
720async def _wait_connected(client: SoloistClient) -> None:
721 """Wait until the client's events WebSocket is connected."""
722 async with asyncio.timeout(5.0):
723 while not client.connected:
724 await asyncio.sleep(0.01)
725
726
727async def test_commands_have_documented_shape(tmp_path: Path) -> None:
728 """Command senders produce the documented wire frames (with clamped values)."""
729 _publish_endpoint(tmp_path)
730 ws = _FakeWebSocket()
731 client = _make_client(tmp_path, ws)
732
733 async def on_event(_event: SoloistEvent) -> None:
734 return
735
736 listen_task = asyncio.create_task(client.listen_events(on_event))
737 await _wait_connected(client)
738
739 await client.play("spotify:playlist:37i9dQZF1DXcBWIGoYBM5M")
740 await client.resume()
741 await client.pause()
742 await client.skip_next()
743 await client.seek(-100)
744 await client.set_volume(150)
745 await client.set_shuffle(True)
746 await client.set_repeat_context(True)
747 await client.add_to_queue("spotify:track:6rqhFgbbKwnb9MLmUQDhG6")
748 await client.get_queue(5)
749 ws.queue.put_nowait(None)
750 await listen_task
751
752 assert ws.sent == [
753 {"type": "command", "command": "play", "uri": "spotify:playlist:37i9dQZF1DXcBWIGoYBM5M"},
754 {"type": "command", "command": "play"},
755 {"type": "command", "command": "pause"},
756 {"type": "command", "command": "skip_next"},
757 {"type": "command", "command": "seek", "position_ms": 0},
758 {"type": "command", "command": "set_volume", "volume": 100},
759 {"type": "command", "command": "set_shuffle", "enabled": True},
760 {"type": "command", "command": "set_repeat_context", "enabled": True},
761 {
762 "type": "command",
763 "command": "add_to_queue",
764 "uri": "spotify:track:6rqhFgbbKwnb9MLmUQDhG6",
765 },
766 {"type": "command", "command": "get_queue", "limit": 5},
767 ]
768
769
770async def test_command_awaits_result(tmp_path: Path) -> None:
771 """A command sent with await_result resolves once its command_result arrives."""
772 _publish_endpoint(tmp_path)
773 ws = _FakeWebSocket()
774 client = _make_client(tmp_path, ws)
775 events: list[SoloistEvent] = []
776
777 async def on_event(event: SoloistEvent) -> None:
778 events.append(event)
779
780 listen_task = asyncio.create_task(client.listen_events(on_event))
781 await _wait_connected(client)
782
783 command_task = asyncio.create_task(client.activate(await_result=True))
784 await asyncio.sleep(0)
785 assert not command_task.done()
786 ws.queue.put_nowait(_text_msg({"type": "command_result", "command": "activate"}))
787 await asyncio.wait_for(command_task, timeout=1.0)
788 ws.queue.put_nowait(None)
789 await listen_task
790
791 # the ack is also still dispatched to the event callback
792 assert [event.type for event in events] == ["command_result"]
793
794
795async def test_commands_require_connection(tmp_path: Path) -> None:
796 """Sending a command without a connected WebSocket raises SoloistError."""
797 client = _make_client(tmp_path, _FakeWebSocket())
798
799 with pytest.raises(SoloistError, match="not connected"):
800 await client.pause()
801
802
803@pytest.mark.usefixtures("linux_platform", "fake_version_cmd")
804async def test_expired_by_timestamp_is_replaced(tmp_path: Path) -> None:
805 """An install past the 90-day build expiry is replaced even though it still runs."""
806 build_a = _build_archive(
807 tmp_path / "a.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-A")}
808 )
809 build_b = _build_archive(
810 tmp_path / "b.tar.gz", {"soloist": _elf_binary("x86_64", marker=b"GOOD-BUILD-B")}
811 )
812 manager, session = _make_manager(tmp_path, _serve_archive(build_a, etag='"v1"'))
813 await manager.ensure_binary(consent=True)
814 _age_metadata(tmp_path, days=100.0)
815 session.handler = _serve_archive(build_b, etag='"v2"')
816
817 path = await manager.ensure_fresh(consent=True)
818
819 assert b"GOOD-BUILD-B" in path.read_bytes()
820