/
/
/
1"""Tests for the TeddyCloud music provider."""
2
3from __future__ import annotations
4
5from typing import Any, Self, cast
6from unittest.mock import AsyncMock, Mock
7
8import pytest
9from music_assistant_models.enums import ContentType, MediaType, StreamType
10from music_assistant_models.streamdetails import StreamDetails
11
12from music_assistant.providers.teddycloud import SUPPORTED_FEATURES
13from music_assistant.providers.teddycloud.constants import TAF_HEADER_SIZE
14from music_assistant.providers.teddycloud.provider import (
15 TeddyCloudProvider,
16 _parse_taf_header,
17 _read_varint,
18)
19
20
21def _encode_varint(value: int) -> bytes:
22 """Encode an int as a protobuf base-128 varint."""
23 out = bytearray()
24 while True:
25 byte = value & 0x7F
26 value >>= 7
27 if value:
28 out.append(byte | 0x80)
29 else:
30 out.append(byte)
31 return bytes(out)
32
33
34def _make_taf_header(num_bytes: int, page_nums: list[int]) -> bytes:
35 """
36 Build a synthetic 4096-byte TAF header.
37
38 Encodes field 2 (num_bytes, varint) and field 4 (track_page_nums, packed varints),
39 prefixed by the 4-byte big-endian protobuf length and zero-padded to the header size.
40 """
41 proto = bytearray()
42 proto += b"\x10" + _encode_varint(num_bytes) # field 2, wire type 0 (varint)
43 packed = b"".join(_encode_varint(p) for p in page_nums)
44 proto += b"\x22" + _encode_varint(len(packed)) + packed # field 4, wire type 2 (packed)
45 header = len(proto).to_bytes(4, "big") + bytes(proto)
46 return header.ljust(TAF_HEADER_SIZE, b"\x00")
47
48
49class _FakeResponse:
50 """Minimal async-context-manager stand-in for an aiohttp response."""
51
52 def __init__(self, data: bytes = b"", json_payload: Any = None) -> None:
53 self._data = data
54 self._json = json_payload
55
56 async def __aenter__(self) -> Self:
57 return self
58
59 async def __aexit__(self, *_exc: object) -> bool:
60 return False
61
62 def raise_for_status(self) -> None:
63 """No-op: the fake response is always OK."""
64
65 async def json(self, content_type: str | None = None) -> Any:
66 return self._json
67
68 @property
69 def content(self) -> Mock:
70 reader = Mock()
71 reader.readexactly = AsyncMock(return_value=self._data)
72 return reader
73
74
75@pytest.fixture
76def mass_mock() -> Mock:
77 """Return a mock MusicAssistant instance."""
78 mass = Mock()
79 mass.http_session = Mock()
80 mass.cache.get = AsyncMock(return_value=None)
81 mass.cache.set = AsyncMock()
82 # setup_data is unset in these unit tests, so get_setup_value falls through to
83 # the provider config's get_value (which the config mock below stubs)
84 mass.config.get = Mock(return_value=None)
85 mass.config.get_raw_provider_config_value = Mock(return_value=None)
86 return mass
87
88
89@pytest.fixture
90def manifest_mock() -> Mock:
91 """Return a mock provider manifest."""
92 manifest = Mock()
93 manifest.domain = "teddycloud"
94 return manifest
95
96
97@pytest.fixture
98def config_mock() -> Mock:
99 """Return a mock provider config."""
100 config = Mock()
101 config.name = "TeddyCloud Test"
102 config.instance_id = "teddycloud_test"
103 config.enabled = True
104 # empty values dict so get_setup_value falls through to config.get_value below
105 config.values = {}
106 # the base provider reads the log level from config on init; return a valid level.
107 config.get_value.side_effect = lambda key, default=None: (
108 "INFO" if key == "log_level" else default
109 )
110 return config
111
112
113@pytest.fixture
114def provider(mass_mock: Mock, manifest_mock: Mock, config_mock: Mock) -> TeddyCloudProvider:
115 """Return a TeddyCloudProvider with the state handle_async_init would set (no network)."""
116 prov = TeddyCloudProvider(mass_mock, manifest_mock, config_mock, SUPPORTED_FEATURES)
117 prov.base_url = "http://teddycloud.local"
118 prov._tags = {}
119 prov._tags_fetched_at = 0.0
120 return prov
121
122
123def _tag(**overrides: Any) -> dict[str, Any]:
124 """Return a representative TeddyCloud tag dict, with optional overrides."""
125 tag: dict[str, Any] = {
126 "uid": "0011223344556677",
127 "tonieInfo": {
128 "series": "Willow Hollow Tales",
129 "episode": "The Copper Lantern",
130 "tracks": ["Story One", "Story Two"],
131 "picture": "https://example.com/pic.png",
132 "language": "de-de",
133 },
134 "trackSeconds": [0, 180, 360, 540],
135 "source": "/library/by/audioID/1600000000.taf",
136 "audioUrl": "/content/download/000000/0011223344556677?skip_header=true",
137 }
138 tag.update(overrides)
139 return tag
140
141
142# --- protobuf / TAF header parsing -------------------------------------------------- #
143
144
145def test_read_varint_roundtrip() -> None:
146 """Encoded varints decode back to their original values."""
147 for value in (0, 1, 127, 128, 300, 16384, 1_615_475_317):
148 decoded, pos = _read_varint(_encode_varint(value), 0)
149 assert decoded == value
150 assert pos == len(_encode_varint(value))
151
152
153def test_parse_taf_header_reads_num_bytes_and_last_page() -> None:
154 """The header parser returns num_bytes (field 2) and the last packed page (field 4)."""
155 header = _make_taf_header(num_bytes=8_192_000, page_nums=[0, 250, 500, 1000])
156 num_bytes, last_page = _parse_taf_header(header)
157 assert num_bytes == 8_192_000
158 assert last_page == 1000
159
160
161def test_parse_taf_header_too_short() -> None:
162 """A truncated header yields (None, None) rather than raising."""
163 assert _parse_taf_header(b"\x00\x00") == (None, None)
164
165
166# --- URL handling ------------------------------------------------------------------- #
167
168
169def test_content_url_strips_skip_header(provider: TeddyCloudProvider) -> None:
170 """_content_url yields the RAW TAF (skip_header removed) so the header can be read."""
171 url = provider._content_url(_tag())
172 assert "skip_header" not in url
173 assert url == "http://teddycloud.local/content/download/000000/0011223344556677"
174
175
176def test_stream_url_adds_skip_header(provider: TeddyCloudProvider) -> None:
177 """_stream_url serves clean OGG/Opus by forcing skip_header=true."""
178 url = provider._stream_url(_tag())
179 assert url.endswith("skip_header=true")
180 assert url.count("skip_header=true") == 1
181
182
183def test_content_url_from_source_without_audiourl(provider: TeddyCloudProvider) -> None:
184 """When audioUrl is absent, the content URL is built from the source path."""
185 tag = _tag(audioUrl=None, source="/library/by/audioID/1600000000.taf")
186 url = provider._content_url(tag)
187 assert url == ("http://teddycloud.local/content/download/library/by/audioID/1600000000.taf")
188
189
190# --- chapter alignment -------------------------------------------------------------- #
191
192
193def test_chapters_one_to_one(provider: TeddyCloudProvider) -> None:
194 """Equal mark/name counts produce one named chapter per mark."""
195 tag = _tag(trackSeconds=[0, 180], tonieInfo={"tracks": ["Story One", "Story Two"]})
196 chapters = provider._chapters(tag, total_seconds=300)
197 assert [c.name for c in chapters] == ["Story One", "Story Two"]
198 assert [c.start for c in chapters] == [0.0, 180.0]
199 assert chapters[-1].end == 300.0
200
201
202def test_chapters_collapse_to_stories(provider: TeddyCloudProvider) -> None:
203 """When marks are an exact multiple of names, collapse to evenly-grouped stories."""
204 tag = _tag(trackSeconds=[0, 180, 360, 540], tonieInfo={"tracks": ["Story One", "Story Two"]})
205 chapters = provider._chapters(tag, total_seconds=720)
206 assert [c.name for c in chapters] == ["Story One", "Story Two"]
207 assert [c.start for c in chapters] == [0.0, 360.0] # group size 2 -> marks 0 and 2
208 assert chapters[-1].end == 720.0
209
210
211def test_chapters_generic_when_unaligned(provider: TeddyCloudProvider) -> None:
212 """A non-divisible mismatch keeps every mark with generic names (no guessed labels)."""
213 tag = _tag(trackSeconds=[0, 180, 360], tonieInfo={"tracks": ["Story One", "Story Two"]})
214 chapters = provider._chapters(tag, total_seconds=None)
215 assert [c.name for c in chapters] == ["Chapter 1", "Chapter 2", "Chapter 3"]
216 assert chapters[-1].end is None # no total -> last chapter unbounded
217
218
219# --- audiobook parsing -------------------------------------------------------------- #
220
221
222def test_parse_audiobook_maps_fields(provider: TeddyCloudProvider) -> None:
223 """A tag maps to an Audiobook with title, series, chapters, duration and date_added."""
224 book = provider._parse_audiobook(_tag(), total_seconds=720)
225 assert book.item_id == "0011223344556677"
226 assert book.name == "The Copper Lantern"
227 assert book.publisher == "Willow Hollow Tales"
228 assert list(book.authors) == ["Willow Hollow Tales"]
229 assert book.duration == 720
230 chapters = book.metadata.chapters
231 assert chapters is not None
232 assert len(chapters) == 2
233 # audio_id 1600000000 is a unix timestamp -> becomes date_added
234 assert book.date_added is not None
235
236
237def test_parse_audiobook_custom_tonie_has_no_publisher(provider: TeddyCloudProvider) -> None:
238 """A tag without a recognised series falls back to the default author, no publisher."""
239 tag = _tag(tonieInfo={"tracks": ["Story One"]}, trackSeconds=[0])
240 book = provider._parse_audiobook(tag, total_seconds=60)
241 assert book.publisher is None
242 assert list(book.authors) == ["TeddyCloud"]
243
244
245# --- duration derivation ------------------------------------------------------------ #
246
247
248async def test_total_seconds_from_header(
249 provider: TeddyCloudProvider, monkeypatch: pytest.MonkeyPatch
250) -> None:
251 """Duration is derived from the TAF byte-rate and persisted to mass.cache."""
252 # byte-rate: 8_192_000 bytes over page 1000 (=1000*4096 bytes) at t=540 -> total 1080
253 header = _make_taf_header(num_bytes=8_192_000, page_nums=[0, 1000])
254 monkeypatch.setattr(provider.mass.http_session, "get", Mock(return_value=_FakeResponse(header)))
255 tag = _tag(trackSeconds=[0, 540]) # audio_id 1600000000 is the cache key
256
257 total = await provider._total_seconds(tag)
258
259 assert total == 1080
260 # the derived value is written to the persistent cache under the audio_id
261 cache_set = cast("AsyncMock", provider.mass.cache.set)
262 cache_set.assert_awaited_once()
263 await_args = cache_set.await_args
264 assert await_args is not None
265 assert await_args.kwargs["key"] == "1600000000"
266 assert await_args.kwargs["data"] == 1080
267
268
269async def test_total_seconds_uses_cache(
270 provider: TeddyCloudProvider, monkeypatch: pytest.MonkeyPatch
271) -> None:
272 """A cached duration is returned without probing the TAF header over HTTP."""
273 monkeypatch.setattr(provider.mass.cache, "get", AsyncMock(return_value=1080))
274 monkeypatch.setattr(
275 provider.mass.http_session, "get", Mock(side_effect=AssertionError("should not fetch"))
276 )
277 assert await provider._total_seconds(_tag(trackSeconds=[0, 540])) == 1080
278
279
280async def test_total_seconds_rejects_impossible_result(
281 provider: TeddyCloudProvider, monkeypatch: pytest.MonkeyPatch
282) -> None:
283 """A derived total shorter than the last chapter start is discarded."""
284 header = _make_taf_header(num_bytes=1000, page_nums=[0, 1000]) # tiny file, big page
285 monkeypatch.setattr(provider.mass.http_session, "get", Mock(return_value=_FakeResponse(header)))
286 tag = _tag(trackSeconds=[0, 540])
287 assert await provider._total_seconds(tag) is None
288
289
290# --- streaming & misc --------------------------------------------------------------- #
291
292
293def test_is_streaming_provider_is_false(provider: TeddyCloudProvider) -> None:
294 """TeddyCloud is a self-hosted library: catalog equals contents."""
295 assert provider.is_streaming_provider is False
296
297
298async def test_get_stream_details(provider: TeddyCloudProvider) -> None:
299 """Stream details report seekable OGG/Opus over HTTP with skip_header applied."""
300 provider._get_tag = AsyncMock(return_value=_tag()) # type: ignore[method-assign]
301 details = await provider.get_stream_details("0011223344556677", MediaType.AUDIOBOOK)
302 assert isinstance(details, StreamDetails)
303 assert details.stream_type == StreamType.HTTP
304 assert details.audio_format.content_type == ContentType.OGG
305 assert details.media_type == MediaType.AUDIOBOOK
306 assert details.can_seek is True
307 assert isinstance(details.path, str)
308 assert details.path.endswith("skip_header=true")
309
310
311async def test_get_library_audiobooks(provider: TeddyCloudProvider) -> None:
312 """The library generator yields one Audiobook per playable tag."""
313 provider._fetch_tags = AsyncMock(return_value={"0011223344556677": _tag()}) # type: ignore[method-assign]
314 provider._total_seconds = AsyncMock(return_value=720) # type: ignore[method-assign]
315 books = [book async for book in provider.get_library_audiobooks()]
316 assert len(books) == 1
317 assert books[0].name == "The Copper Lantern"
318
319
320async def test_fetch_tags_normalizes_ruid_only(
321 provider: TeddyCloudProvider, monkeypatch: pytest.MonkeyPatch
322) -> None:
323 """A tag with only 'ruid' (no 'uid') is normalized so downstream tag['uid'] is safe."""
324 payload = {
325 "tags": [
326 {
327 "ruid": "AABBCCDD00112233",
328 "trackSeconds": [0],
329 "source": "/library/by/audioID/1600000000.taf",
330 "audioUrl": "/content/download/000000/AABBCCDD00112233",
331 }
332 ]
333 }
334 monkeypatch.setattr(
335 provider.mass.http_session, "get", Mock(return_value=_FakeResponse(json_payload=payload))
336 )
337
338 tags = await provider._fetch_tags(force=True)
339
340 assert "AABBCCDD00112233" in tags
341 assert tags["AABBCCDD00112233"]["uid"] == "AABBCCDD00112233" # normalized from ruid
342 # parsing must not raise KeyError on tag["uid"]
343 book = provider._parse_audiobook(tags["AABBCCDD00112233"], total_seconds=None)
344 assert book.item_id == "AABBCCDD00112233"
345
346
347async def test_handle_async_init_prepends_scheme(
348 mass_mock: Mock, manifest_mock: Mock, monkeypatch: pytest.MonkeyPatch
349) -> None:
350 """A server entered as a bare host/IP is normalized to an http:// base URL."""
351 config = Mock()
352 config.instance_id = "teddycloud_test"
353 config.values = {}
354 values: dict[str, Any] = {"url": "192.168.3.225", "log_level": "INFO"}
355 config.get_value.side_effect = lambda key, default=None: values.get(key, default)
356 prov = TeddyCloudProvider(mass_mock, manifest_mock, config, SUPPORTED_FEATURES)
357 monkeypatch.setattr(
358 mass_mock.http_session, "get", Mock(return_value=_FakeResponse(json_payload={"tags": []}))
359 )
360
361 await prov.handle_async_init()
362
363 assert prov.base_url == "http://192.168.3.225"
364