/
/
/
1"""Test that seeking a Deezer track starts the stream where it was asked to."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Self
6from unittest.mock import Mock
7
8import pytest
9from music_assistant_models.enums import ContentType
10from music_assistant_models.media_items import AudioFormat
11from music_assistant_models.streamdetails import StreamDetails
12
13from music_assistant.providers.deezer.streaming import DeezerStreamingManager
14
15if TYPE_CHECKING:
16 from collections.abc import AsyncGenerator
17
18CHUNK_SIZE = 2048
19DURATION = 600
20
21
22class _Track:
23 """A fake encrypted track, every chunk carries its own index as payload."""
24
25 status = 200
26
27 def __init__(self, chunk_count: int) -> None:
28 self._data = b"".join(
29 f"{index:08d}".encode() * (CHUNK_SIZE // 8) for index in range(chunk_count)
30 )
31
32 async def _iter_chunked(self, size: int) -> AsyncGenerator[bytes]:
33 for start in range(0, len(self._data), size):
34 yield self._data[start : start + size]
35
36 @property
37 def content(self) -> Mock:
38 return Mock(iter_chunked=self._iter_chunked)
39
40 async def __aenter__(self) -> Self:
41 return self
42
43 async def __aexit__(self, *args: object) -> bool:
44 return False
45
46
47async def _played_from(
48 monkeypatch: pytest.MonkeyPatch, bitrate_kbit: int, seek_position: int
49) -> float:
50 """Return the position in seconds the stream actually starts playing at."""
51 bytes_per_second = bitrate_kbit * 1000 // 8
52 size = bytes_per_second * DURATION
53
54 provider = Mock()
55 provider.mass.http_session.get = lambda *_args, **_kwargs: _Track(size // CHUNK_SIZE)
56 streaming = DeezerStreamingManager(provider)
57 monkeypatch.setattr(streaming, "_get_blowfish_key", lambda _track_id: "0" * 16)
58 monkeypatch.setattr(streaming, "_decrypt_chunk", lambda chunk, _key: chunk)
59
60 streamdetails = StreamDetails(
61 provider="deezer--test",
62 item_id="1",
63 audio_format=AudioFormat(content_type=ContentType.MP3),
64 duration=DURATION,
65 size=size,
66 data={"track_id": "1", "url": "http://x"},
67 )
68 chunks = [c async for c in streaming._stream_encrypted_track(streamdetails, seek_position)]
69 # the first chunk is always sent, playback continues from the one after it
70 return int(chunks[1][:8]) * CHUNK_SIZE / bytes_per_second
71
72
73@pytest.mark.parametrize("bitrate_kbit", [128, 320, 800])
74@pytest.mark.parametrize("seek_position", [30, 60, 300])
75async def test_seek_starts_within_one_chunk(
76 monkeypatch: pytest.MonkeyPatch, bitrate_kbit: int, seek_position: int
77) -> None:
78 """A seek may only be off by the chunk it cannot split."""
79 chunk_seconds = CHUNK_SIZE / (bitrate_kbit * 1000 / 8)
80 played_from = await _played_from(monkeypatch, bitrate_kbit, seek_position)
81
82 assert abs(played_from - seek_position) <= chunk_seconds
83
84
85async def test_drift_does_not_grow_with_the_seek_position(
86 monkeypatch: pytest.MonkeyPatch,
87) -> None:
88 """Truncating the chunks per second first scaled the drift with the distance."""
89 near = 60 - await _played_from(monkeypatch, 128, 60)
90 far = 300 - await _played_from(monkeypatch, 128, 300)
91
92 assert far == pytest.approx(near, abs=CHUNK_SIZE / (128 * 1000 / 8))
93