/
/
/
1"""Tests for playlist parsing and generation helpers."""
2
3from types import TracebackType
4from typing import Any, Self, cast
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8from aiohttp import client_exceptions
9from music_assistant_models.enums import ContentType, ExternalID, ImageType, MediaType
10from music_assistant_models.errors import InvalidDataError
11from music_assistant_models.media_items import (
12 AudioFormat,
13 ItemMapping,
14 MediaItemImage,
15 MediaItemMetadata,
16 ProviderMapping,
17 Radio,
18 SoundEffect,
19 Track,
20 UniqueList,
21)
22
23from music_assistant.controllers.music.media.playlists import PlaylistController
24from music_assistant.helpers import playlists
25from music_assistant.helpers.playlists import (
26 AlbumInfo,
27 ArtistInfo,
28 ImageInfo,
29 IsHLSPlaylist,
30 PlaylistItem,
31 ProviderMappingInfo,
32 construct_media_item_from_playlist_item,
33 fetch_playlist,
34 generate_m3u,
35 media_item_to_playlist_item,
36 parse_extinf_title,
37 parse_m3u,
38 parse_m3u_playlist_image,
39 parse_m3u_playlist_name,
40 sanitize_m3u_value,
41)
42
43# --------------------------------------------------------------------------- #
44# Existing tests (EXTINF parsing) #
45# --------------------------------------------------------------------------- #
46
47
48def test_m3u_extinf_duration_not_truncated() -> None:
49 """Test that EXTINF duration is parsed as full string, not truncated to first char."""
50 m3u_data = "#EXTM3U\n#EXTINF:120,Test Song\nhttp://example.com/song.mp3\n"
51 result = parse_m3u(m3u_data)
52 assert len(result) == 1
53 assert result[0].length == "120"
54 assert result[0].title == "Test Song"
55
56
57def test_m3u_extinf_negative_duration() -> None:
58 """Test that EXTINF with -1 duration is treated as None (unknown length)."""
59 m3u_data = "#EXTM3U\n#EXTINF:-1,Live Stream\nhttp://example.com/stream\n"
60 result = parse_m3u(m3u_data)
61 assert len(result) == 1
62 assert result[0].length is None
63 assert result[0].title == "Live Stream"
64
65
66def test_m3u_extinf_single_digit_duration() -> None:
67 """Test that single-digit durations still work correctly."""
68 m3u_data = "#EXTM3U\n#EXTINF:5,Short Clip\nhttp://example.com/clip.mp3\n"
69 result = parse_m3u(m3u_data)
70 assert len(result) == 1
71 assert result[0].length == "5"
72
73
74# --------------------------------------------------------------------------- #
75# parse_extinf_title #
76# --------------------------------------------------------------------------- #
77
78
79def test_parse_extinf_title_with_artist() -> None:
80 """Test parsing 'Artist - Title' format."""
81 artist, title = parse_extinf_title("Radiohead - Everything In Its Right Place")
82 assert artist == "Radiohead"
83 assert title == "Everything In Its Right Place"
84
85
86def test_parse_extinf_title_without_artist() -> None:
87 """Test parsing title without artist separator."""
88 artist, title = parse_extinf_title("Just A Title")
89 assert artist is None
90 assert title == "Just A Title"
91
92
93def test_parse_extinf_title_none() -> None:
94 """Test parsing None title."""
95 artist, title = parse_extinf_title(None)
96 assert artist is None
97 assert title is None
98
99
100def test_parse_extinf_title_multiple_separators() -> None:
101 """Test that only the first ' - ' is used as separator."""
102 artist, title = parse_extinf_title("Artist - Title - Remix")
103 assert artist == "Artist"
104 assert title == "Title - Remix"
105
106
107# --------------------------------------------------------------------------- #
108# EXTMA metadata parsing #
109# --------------------------------------------------------------------------- #
110
111
112def test_m3u_extma_parsing() -> None:
113 """Test that #EXTMA metadata is parsed into the metadata dict."""
114 m3u_data = (
115 "#EXTM3U\n"
116 "#EXTMA:media_type=track||isrc=USRC17607839||album=OK Computer\n"
117 "#EXTINF:240,Radiohead - Everything In Its Right Place\n"
118 "spotify://track/abc123\n"
119 )
120 result = parse_m3u(m3u_data)
121 assert len(result) == 1
122 assert result[0].metadata is not None
123 assert result[0].metadata["media_type"] == "track"
124 assert result[0].metadata["isrc"] == "USRC17607839"
125 assert result[0].metadata["album"] == "OK Computer"
126
127
128def test_m3u_extma_without_metadata() -> None:
129 """Test that entries without EXTMA have None metadata."""
130 m3u_data = "#EXTM3U\n#EXTINF:120,Test\nhttp://example.com/song.mp3\n"
131 result = parse_m3u(m3u_data)
132 assert result[0].metadata is None
133
134
135# --------------------------------------------------------------------------- #
136# EXTPROV provider mapping parsing #
137# --------------------------------------------------------------------------- #
138
139
140def test_m3u_extprov_parsing() -> None:
141 """Test that #EXTPROV lines are parsed into provider mappings."""
142 m3u_data = (
143 "#EXTM3U\n"
144 "#EXTPROV:spotify||abc123||spotify_1||flac||96000||24||320\n"
145 "#EXTPROV:tidal||xyz789||tidal_1||flac||192000||24||0\n"
146 "#EXTINF:240,Radiohead - Everything In Its Right Place\n"
147 "spotify://track/abc123\n"
148 )
149 result = parse_m3u(m3u_data)
150 assert len(result) == 1
151 assert len(result[0].providers) == 2
152 prov1 = result[0].providers[0]
153 assert prov1.domain == "spotify"
154 assert prov1.item_id == "abc123"
155 assert prov1.instance_id == "spotify_1"
156 assert prov1.content_type == "flac"
157 assert prov1.sample_rate == 96000
158 assert prov1.bit_depth == 24
159 assert prov1.bit_rate == 320
160 prov2 = result[0].providers[1]
161 assert prov2.domain == "tidal"
162 assert prov2.item_id == "xyz789"
163 assert prov2.instance_id == "tidal_1"
164 assert prov2.sample_rate == 192000
165
166
167def test_m3u_extprov_minimal() -> None:
168 """Test EXTPROV with only the 2 required fields (domain and item_id)."""
169 m3u_data = "#EXTM3U\n#EXTPROV:spotify||abc123\n#EXTINF:120,Test\nspotify://track/abc123\n"
170 result = parse_m3u(m3u_data)
171 assert len(result[0].providers) == 1
172 assert result[0].providers[0].domain == "spotify"
173 assert result[0].providers[0].item_id == "abc123"
174 assert result[0].providers[0].instance_id == ""
175 assert result[0].providers[0].sample_rate == 0
176
177
178def test_m3u_extprov_invalid_skipped() -> None:
179 """Test that malformed EXTPROV lines are skipped."""
180 m3u_data = "#EXTM3U\n#EXTPROV:onlyonefield\n#EXTINF:120,Test\nhttp://example.com/song.mp3\n"
181 result = parse_m3u(m3u_data)
182 assert len(result[0].providers) == 0
183
184
185# --------------------------------------------------------------------------- #
186# EXTIMG image parsing #
187# --------------------------------------------------------------------------- #
188
189
190def test_m3u_extimg_parsing() -> None:
191 """Test that #EXTIMG lines are parsed into image info."""
192 m3u_data = (
193 "#EXTM3U\n"
194 "#EXTIMG:thumb||https://img.example.com/abc.jpg||spotify||true\n"
195 "#EXTINF:120,Test\n"
196 "spotify://track/abc123\n"
197 )
198 result = parse_m3u(m3u_data)
199 assert len(result[0].images) == 1
200 img = result[0].images[0]
201 assert img.type == "thumb"
202 assert img.path == "https://img.example.com/abc.jpg"
203 assert img.provider == "spotify"
204 assert img.remotely_accessible is True
205
206
207def test_m3u_extimg_not_remotely_accessible() -> None:
208 """Test EXTIMG with remotely_accessible=false."""
209 m3u_data = "#EXTM3U\n#EXTIMG:thumb||/local/path.jpg||builtin||false\n#EXTINF:120,Test\ntest\n"
210 result = parse_m3u(m3u_data)
211 assert result[0].images[0].remotely_accessible is False
212
213
214def test_m3u_extimg_multiple() -> None:
215 """Test multiple EXTIMG lines per entry."""
216 m3u_data = (
217 "#EXTM3U\n"
218 "#EXTIMG:thumb||https://thumb.jpg||spotify||true\n"
219 "#EXTIMG:fanart||https://fanart.jpg||spotify||true\n"
220 "#EXTINF:120,Test\n"
221 "spotify://track/abc123\n"
222 )
223 result = parse_m3u(m3u_data)
224 assert len(result[0].images) == 2
225
226
227# --------------------------------------------------------------------------- #
228# #PLAYLIST directive #
229# --------------------------------------------------------------------------- #
230
231
232def test_parse_m3u_playlist_name() -> None:
233 """Test extracting playlist name from #PLAYLIST directive."""
234 m3u_data = "#EXTM3U\n#PLAYLIST:My Playlist\n#EXTINF:120,Test\ntest.mp3\n"
235 assert parse_m3u_playlist_name(m3u_data) == "My Playlist"
236
237
238def test_parse_m3u_playlist_name_missing() -> None:
239 """Test that None is returned when no #PLAYLIST directive exists."""
240 m3u_data = "#EXTM3U\n#EXTINF:120,Test\ntest.mp3\n"
241 assert parse_m3u_playlist_name(m3u_data) is None
242
243
244def test_parse_m3u_playlist_image() -> None:
245 """Test extracting a playlist-level cover image from #EXTIMG."""
246 m3u_data = (
247 "#EXTM3U\n"
248 "#EXTIMG:https://img.example.com/cover.jpg\n"
249 "#PLAYLIST:My Playlist\n"
250 "#EXTINF:120,Test\n"
251 "test.mp3\n"
252 )
253 assert parse_m3u_playlist_image(m3u_data) == "https://img.example.com/cover.jpg"
254
255
256def test_parse_m3u_playlist_image_missing() -> None:
257 """Test that None is returned when no playlist-level image exists."""
258 m3u_data = "#EXTM3U\n#PLAYLIST:My Playlist\n#EXTINF:120,Test\ntest.mp3\n"
259 assert parse_m3u_playlist_image(m3u_data) is None
260
261
262# --------------------------------------------------------------------------- #
263# generate_m3u #
264# --------------------------------------------------------------------------- #
265
266
267def test_generate_m3u_basic() -> None:
268 """Test basic M3U generation with EXTINF."""
269 items = [
270 PlaylistItem(path="spotify://track/abc123", title="Artist - Song", length="240"),
271 ]
272 result = generate_m3u("My Playlist", items)
273 assert "#EXTM3U\n" in result
274 assert "#PLAYLIST:My Playlist\n" in result
275 assert "#EXTINF:240,Artist - Song\n" in result
276 assert "spotify://track/abc123\n" in result
277
278
279def test_generate_m3u_with_metadata() -> None:
280 """Test M3U generation with EXTMA metadata."""
281 items = [
282 PlaylistItem(
283 path="spotify://track/abc123",
284 title="Artist - Song",
285 length="240",
286 metadata={"media_type": "track", "isrc": "USRC123"},
287 ),
288 ]
289 result = generate_m3u("Test", items)
290 assert "#EXTMA:media_type=track||isrc=USRC123\n" in result
291
292
293def test_generate_m3u_with_providers() -> None:
294 """Test M3U generation with EXTPROV lines."""
295 items = [
296 PlaylistItem(
297 path="spotify://track/abc123",
298 title="Test",
299 length="120",
300 providers=[
301 ProviderMappingInfo(
302 domain="spotify",
303 item_id="abc123",
304 instance_id="spotify_1",
305 content_type="flac",
306 sample_rate=96000,
307 bit_depth=24,
308 bit_rate=320,
309 ),
310 ],
311 ),
312 ]
313 result = generate_m3u("Test", items)
314 assert "#EXTPROV:spotify||abc123||spotify_1||flac||96000||24||320\n" in result
315
316
317def test_generate_m3u_with_images() -> None:
318 """Test M3U generation with EXTIMG lines."""
319 items = [
320 PlaylistItem(
321 path="spotify://track/abc123",
322 title="Test",
323 length="120",
324 images=[
325 ImageInfo(
326 type="thumb",
327 path="https://img.jpg",
328 provider="spotify",
329 remotely_accessible=True,
330 )
331 ],
332 ),
333 ]
334 result = generate_m3u("Test", items)
335 assert "#EXTIMG:thumb||https://img.jpg||spotify||true\n" in result
336
337
338def test_generate_m3u_with_playlist_image() -> None:
339 """Test M3U generation with a playlist-level cover image."""
340 items = [PlaylistItem(path="spotify://track/abc123", title="Test", length="120")]
341 result = generate_m3u("Test", items, "https://img.example.com/cover.jpg")
342 assert result.startswith("#EXTM3U\n#EXTIMG:https://img.example.com/cover.jpg\n#PLAYLIST:Test\n")
343
344
345def test_generate_m3u_empty() -> None:
346 """Test generating an empty M3U playlist."""
347 result = generate_m3u("Empty Playlist", [])
348 assert result == "#EXTM3U\n#PLAYLIST:Empty Playlist\n"
349
350
351@pytest.mark.parametrize(
352 "line_break",
353 ["\n", "\r\n", "\r", "\v", "\f", "\x1c", "\x1d", "\x1e", "\x85", "\u2028", "\u2029"],
354)
355def test_generate_m3u_never_splits_an_entry_over_multiple_lines(line_break: str) -> None:
356 """Test that a line break inside a value cannot split an entry into a second entry."""
357 items = [
358 PlaylistItem(
359 path="spotify://track/abc123",
360 length="240",
361 title=f"Artist - Song (feat. Mad{line_break}elyn Brown)",
362 metadata={"media_type": "track", "name": f"Song{line_break}X"},
363 providers=[ProviderMappingInfo(domain="spotify", item_id="abc123")],
364 images=[
365 ImageInfo(type="thumb", path=f"https://img{line_break}.jpg", provider="spotify")
366 ],
367 artists=[
368 ArtistInfo(
369 name=f"Mad{line_break}elyn Brown",
370 provider_domain="spotify",
371 item_id="art1",
372 provider_instance="spotify",
373 )
374 ],
375 album=AlbumInfo(
376 name=f"The{line_break}Album",
377 provider_domain="spotify",
378 item_id="alb1",
379 provider_instance="spotify",
380 ),
381 ),
382 ]
383 result = generate_m3u(f"My{line_break}Playlist", items, f"https://cover{line_break}.jpg")
384
385 parsed = parse_m3u(result)
386 assert len(parsed) == 1
387 assert parsed[0].path == "spotify://track/abc123"
388 assert parse_m3u_playlist_name(result) == "My Playlist".replace(" ", " " * len(line_break))
389
390
391def test_generate_m3u_leaves_values_without_line_breaks_alone() -> None:
392 """Test that sanitizing only touches line breaks."""
393 title = "Artist - Song (feat. Madelyn Brown)"
394 assert sanitize_m3u_value(title) == title
395 items = [PlaylistItem(path="spotify://track/abc123", title=title, length="240")]
396 assert f"#EXTINF:240,{title}\n" in generate_m3u("My Playlist", items)
397
398
399def test_generate_m3u_no_extinf_without_title() -> None:
400 """Test that entries without title/length skip the EXTINF line."""
401 items = [PlaylistItem(path="spotify://track/abc123")]
402 result = generate_m3u("Test", items)
403 assert "#EXTINF" not in result
404 assert "spotify://track/abc123\n" in result
405
406
407# --------------------------------------------------------------------------- #
408# Round-trip: generate -> parse #
409# --------------------------------------------------------------------------- #
410
411
412def test_round_trip_full() -> None:
413 """Test that generate_m3u output can be parsed back with all metadata preserved."""
414 original = PlaylistItem(
415 path="spotify://track/abc123",
416 title="Radiohead - Everything In Its Right Place",
417 length="240",
418 metadata={"media_type": "track", "isrc": "USRC17607839", "album": "OK Computer"},
419 providers=[
420 ProviderMappingInfo(
421 domain="spotify",
422 item_id="abc123",
423 instance_id="spotify_1",
424 content_type="flac",
425 sample_rate=96000,
426 bit_depth=24,
427 bit_rate=320,
428 ),
429 ProviderMappingInfo(
430 domain="tidal",
431 item_id="xyz789",
432 instance_id="tidal_1",
433 content_type="flac",
434 sample_rate=192000,
435 bit_depth=24,
436 bit_rate=0,
437 ),
438 ],
439 images=[
440 ImageInfo(
441 type="thumb",
442 path="https://img.example.com/thumb.jpg",
443 provider="spotify",
444 remotely_accessible=True,
445 ),
446 ],
447 )
448
449 m3u_data = generate_m3u("Test Playlist", [original])
450 parsed = parse_m3u(m3u_data)
451
452 assert len(parsed) == 1
453 item = parsed[0]
454 assert item.path == original.path
455 assert item.title == original.title
456 assert item.length == original.length
457
458 # metadata round-trip
459 assert item.metadata == original.metadata
460
461 # provider round-trip
462 assert len(item.providers) == 2
463 assert item.providers[0].domain == "spotify"
464 assert item.providers[0].item_id == "abc123"
465 assert item.providers[0].instance_id == "spotify_1"
466 assert item.providers[0].content_type == "flac"
467 assert item.providers[0].sample_rate == 96000
468 assert item.providers[0].bit_depth == 24
469 assert item.providers[0].bit_rate == 320
470 assert item.providers[1].domain == "tidal"
471 assert item.providers[1].sample_rate == 192000
472
473 # image round-trip
474 assert len(item.images) == 1
475 assert item.images[0].type == "thumb"
476 assert item.images[0].path == "https://img.example.com/thumb.jpg"
477 assert item.images[0].provider == "spotify"
478 assert item.images[0].remotely_accessible is True
479
480 # playlist name round-trip
481 assert parse_m3u_playlist_name(m3u_data) == "Test Playlist"
482
483
484def test_round_trip_multiple_entries() -> None:
485 """Test round-trip with multiple entries preserves order and all data."""
486 items = [
487 PlaylistItem(
488 path="spotify://track/track1",
489 title="Artist A - Song 1",
490 length="200",
491 metadata={"media_type": "track"},
492 ),
493 PlaylistItem(
494 path="tidal://track/track2",
495 title="Artist B - Song 2",
496 length="300",
497 metadata={"media_type": "track"},
498 ),
499 PlaylistItem(
500 path="builtin://radio/http://stream.example.com",
501 title="Radio Station",
502 length="-1", # will be parsed as None
503 metadata={"media_type": "radio"},
504 ),
505 ]
506 m3u_data = generate_m3u("Multi", items)
507 parsed = parse_m3u(m3u_data)
508
509 assert len(parsed) == 3
510 assert parsed[0].path == "spotify://track/track1"
511 assert parsed[1].path == "tidal://track/track2"
512 assert parsed[2].path == "builtin://radio/http://stream.example.com"
513 # -1 duration is normalized to None on parse
514 assert parsed[2].length is None
515
516
517def test_round_trip_unknown_length_keeps_title() -> None:
518 """Test that an entry with unknown length survives repeated rewrites."""
519 original = PlaylistItem(
520 path="builtin://radio/http://stream.example.com",
521 title="My Custom Station Name",
522 metadata={"media_type": "radio", "name": "My Custom Station Name"},
523 images=[
524 ImageInfo(
525 type="thumb",
526 path="https://img.example.com/station.jpg",
527 provider="builtin",
528 remotely_accessible=True,
529 ),
530 ],
531 )
532 # playlist files are rewritten on every edit, so parse -> generate must be lossless
533 parsed = parse_m3u(generate_m3u("Radio", [original]))
534 reparsed = parse_m3u(generate_m3u("Radio", parsed))
535
536 assert len(reparsed) == 1
537 assert reparsed[0].title == "My Custom Station Name"
538 assert reparsed[0].length is None
539 assert reparsed[0].metadata == original.metadata
540 assert reparsed[0].images[0].path == "https://img.example.com/station.jpg"
541
542
543def test_round_trip_bare_uris() -> None:
544 """Test round-trip with bare URIs (no metadata) - migrated playlists."""
545 items = [
546 PlaylistItem(path="spotify://track/abc123"),
547 PlaylistItem(path="tidal://track/xyz789"),
548 ]
549 m3u_data = generate_m3u("Migrated", items)
550 parsed = parse_m3u(m3u_data)
551 assert len(parsed) == 2
552 assert parsed[0].path == "spotify://track/abc123"
553 assert parsed[0].metadata is None
554 assert parsed[0].providers == []
555 assert parsed[1].path == "tidal://track/xyz789"
556
557
558def test_construct_media_item_from_playlist_item_sound_effect() -> None:
559 """Stored playlist metadata reconstructs a SoundEffect with mappings and artwork."""
560
561 class DummyProvider:
562 def __init__(self, domain: str, instance_id: str) -> None:
563 self.domain = domain
564 self.instance_id = instance_id
565
566 mass = MagicMock()
567 builtin_provider = DummyProvider("builtin", "builtin_1")
568 mass.get_provider.side_effect = lambda ref: {
569 "builtin": builtin_provider,
570 "builtin_1": builtin_provider,
571 }.get(ref)
572 item = PlaylistItem(
573 path="builtin://sound_effect/http://example.com/chime.mp3",
574 title="Chime",
575 length="12",
576 metadata={
577 "media_type": MediaType.SOUND_EFFECT.value,
578 "name": "Chime",
579 "mbid": "soundeffect-mbid",
580 },
581 providers=[
582 ProviderMappingInfo(
583 domain="builtin",
584 item_id="http://example.com/chime.mp3",
585 instance_id="builtin_1",
586 content_type="mp3",
587 sample_rate=44100,
588 bit_depth=16,
589 bit_rate=192,
590 )
591 ],
592 images=[
593 ImageInfo(
594 type="thumb",
595 path="https://example.com/chime.jpg",
596 provider="builtin",
597 remotely_accessible=True,
598 )
599 ],
600 )
601
602 result = construct_media_item_from_playlist_item(item, mass)
603
604 assert isinstance(result, SoundEffect)
605 assert result.name == "Chime"
606 assert result.duration == 12
607 assert result.get_external_id(ExternalID.MB_RECORDING) == "soundeffect-mbid"
608 assert result.provider_mappings
609 mapping = next(iter(result.provider_mappings))
610 assert mapping.provider_domain == "builtin"
611 assert mapping.provider_instance == "builtin_1"
612 assert mapping.item_id == "http://example.com/chime.mp3"
613 assert result.metadata.images
614 assert result.metadata.images[0].path == "https://example.com/chime.jpg"
615
616
617def test_media_item_to_playlist_item_sound_effect_round_trip() -> None:
618 """SoundEffect playlist items round-trip through M3U metadata without losing data."""
619
620 class DummyProvider:
621 def __init__(self, domain: str, instance_id: str) -> None:
622 self.domain = domain
623 self.instance_id = instance_id
624
625 mass = MagicMock()
626 builtin_provider = DummyProvider("builtin", "builtin_1")
627 mass.get_provider.side_effect = lambda ref: {
628 "builtin": builtin_provider,
629 "builtin_1": builtin_provider,
630 }.get(ref)
631 image = MediaItemImage(
632 type=ImageType.THUMB,
633 path="https://example.com/chime.jpg",
634 provider="builtin",
635 remotely_accessible=True,
636 )
637 sound_effect = SoundEffect(
638 item_id="http://example.com/chime.mp3",
639 provider="builtin",
640 name="Chime",
641 provider_mappings={
642 ProviderMapping(
643 item_id="http://example.com/chime.mp3",
644 provider_domain="builtin",
645 provider_instance="builtin_1",
646 audio_format=AudioFormat(
647 content_type=ContentType.MP3,
648 sample_rate=44100,
649 bit_depth=16,
650 bit_rate=192,
651 ),
652 )
653 },
654 external_ids={(ExternalID.MB_RECORDING, "soundeffect-mbid")},
655 metadata=MediaItemMetadata(images=UniqueList([image])),
656 )
657 sound_effect.duration = 12
658
659 playlist_item = media_item_to_playlist_item(sound_effect)
660 parsed = parse_m3u(generate_m3u("FX", [playlist_item]))
661 reconstructed = construct_media_item_from_playlist_item(parsed[0], mass)
662
663 assert playlist_item.path == "builtin://sound_effect/http://example.com/chime.mp3"
664 assert playlist_item.metadata == {
665 "media_type": MediaType.SOUND_EFFECT.value,
666 "name": "Chime",
667 "mbid": "soundeffect-mbid",
668 }
669 assert playlist_item.length == "12"
670 assert playlist_item.images[0].path == "https://example.com/chime.jpg"
671 assert isinstance(reconstructed, SoundEffect)
672 assert reconstructed.duration == 12
673 assert reconstructed.provider_mappings
674 mapping = next(iter(reconstructed.provider_mappings))
675 assert mapping.provider_instance == "builtin_1"
676 assert mapping.item_id == "http://example.com/chime.mp3"
677 assert reconstructed.metadata.images
678 assert reconstructed.metadata.images[0].path == "https://example.com/chime.jpg"
679
680
681# --------------------------------------------------------------------------- #
682# media_item_to_playlist_item tests #
683# --------------------------------------------------------------------------- #
684
685
686def test_media_item_to_playlist_item_track() -> None:
687 """Test conversion of a Track with full metadata to PlaylistItem."""
688 artist = ItemMapping(
689 item_id="art1", provider="spotify", name="Radiohead", media_type=MediaType.ARTIST
690 )
691 album = ItemMapping(
692 item_id="alb1", provider="spotify", name="Kid A", media_type=MediaType.ALBUM
693 )
694 img = MediaItemImage(
695 type=ImageType.THUMB,
696 path="https://example.com/img.jpg",
697 provider="spotify",
698 remotely_accessible=True,
699 )
700 track = Track(
701 item_id="abc123",
702 provider="spotify",
703 name="Everything In Its Right Place",
704 duration=240,
705 version="Deluxe",
706 provider_mappings={
707 ProviderMapping(
708 item_id="abc123",
709 provider_domain="spotify",
710 provider_instance="spotify_1",
711 audio_format=AudioFormat(
712 content_type=ContentType.FLAC,
713 sample_rate=44100,
714 bit_depth=16,
715 bit_rate=320,
716 ),
717 ),
718 },
719 artists=UniqueList([artist]),
720 album=album,
721 external_ids={(ExternalID.ISRC, "USRC17607839")},
722 metadata=MediaItemMetadata(images=UniqueList([img])),
723 )
724
725 result = media_item_to_playlist_item(track)
726
727 assert result.path == "spotify://track/abc123"
728 assert result.title == "Radiohead - Everything In Its Right Place"
729 assert result.length == "240"
730 assert result.metadata is not None
731 assert result.metadata["media_type"] == "track"
732 assert result.metadata["name"] == "Everything In Its Right Place"
733 assert result.metadata["isrc"] == "USRC17607839"
734 assert result.metadata["version"] == "Deluxe"
735 assert len(result.providers) == 1
736 assert result.providers[0].domain == "spotify"
737 assert result.providers[0].item_id == "abc123"
738 assert result.providers[0].content_type == "flac"
739 assert result.providers[0].sample_rate == 44100
740 assert result.providers[0].bit_rate == 320
741 assert len(result.artists) == 1
742 assert result.artists[0].name == "Radiohead"
743 assert result.album is not None
744 assert result.album.name == "Kid A"
745 assert len(result.images) == 1
746 assert result.images[0].type == "thumb"
747 assert result.images[0].remotely_accessible is True
748
749
750def test_media_item_to_playlist_item_radio() -> None:
751 """Test conversion of a Radio to PlaylistItem."""
752 radio = Radio(
753 item_id="radio1",
754 provider="builtin",
755 name="Test FM",
756 provider_mappings={
757 ProviderMapping(
758 item_id="radio1",
759 provider_domain="builtin",
760 provider_instance="builtin",
761 audio_format=AudioFormat(content_type=ContentType.OGG),
762 ),
763 },
764 )
765
766 result = media_item_to_playlist_item(radio)
767
768 assert result.path == "builtin://radio/radio1"
769 assert result.title == "Test FM"
770 assert result.metadata is not None
771 assert result.metadata["media_type"] == "radio"
772 assert result.podcast is None
773 assert result.album is None
774 assert len(result.artists) == 0
775 # a radio has no duration, which must still yield an #EXTINF line carrying the name
776 assert result.length is None
777 assert "#EXTINF:-1,Test FM" in generate_m3u("Radio Stations", [result])
778
779
780def test_media_item_to_playlist_item_no_version() -> None:
781 """Test that version is omitted from metadata when empty."""
782 track = Track(
783 item_id="t1",
784 provider="tidal",
785 name="Simple Track",
786 duration=180,
787 provider_mappings={
788 ProviderMapping(
789 item_id="t1",
790 provider_domain="tidal",
791 provider_instance="tidal_1",
792 audio_format=AudioFormat(content_type=ContentType.FLAC),
793 ),
794 },
795 )
796
797 result = media_item_to_playlist_item(track)
798
799 assert result.metadata is not None
800 assert "version" not in result.metadata
801
802
803def test_media_item_to_playlist_item_multiple_providers() -> None:
804 """Test that multiple provider mappings are collected, one per domain, highest quality first."""
805 track = Track(
806 item_id="t1",
807 provider="spotify",
808 name="Multi Provider Track",
809 duration=200,
810 provider_mappings={
811 ProviderMapping(
812 item_id="t1",
813 provider_domain="spotify",
814 provider_instance="spotify_1",
815 audio_format=AudioFormat(
816 content_type=ContentType.OGG, sample_rate=44100, bit_depth=0, bit_rate=320
817 ),
818 ),
819 ProviderMapping(
820 item_id="t2",
821 provider_domain="tidal",
822 provider_instance="tidal_1",
823 audio_format=AudioFormat(
824 content_type=ContentType.FLAC, sample_rate=96000, bit_depth=24, bit_rate=0
825 ),
826 ),
827 },
828 )
829
830 result = media_item_to_playlist_item(track)
831
832 assert len(result.providers) == 2
833 domains = {p.domain for p in result.providers}
834 assert domains == {"spotify", "tidal"}
835 # primary URI uses the highest quality provider
836 assert result.path == "tidal://track/t2"
837
838
839def test_radio_entries_round_trip() -> None:
840 """Test that radio entries survive generation and parsing unchanged."""
841 items = [
842 PlaylistItem(
843 path="http://stream.example.com/radio1",
844 title="Jazz FM",
845 length="-1",
846 metadata={"media_type": "radio"},
847 ),
848 PlaylistItem(
849 path="http://stream.example.com/radio2",
850 title="Classic Rock Radio",
851 length="-1",
852 metadata={"media_type": "radio"},
853 ),
854 ]
855 m3u_data = generate_m3u("Radio Stations", items)
856 parsed = parse_m3u(m3u_data)
857
858 assert len(parsed) == 2
859 assert parsed[0].path == "http://stream.example.com/radio1"
860 assert parsed[0].title == "Jazz FM"
861 assert parsed[0].metadata is not None
862 assert parsed[0].metadata["media_type"] == "radio"
863 assert parsed[1].path == "http://stream.example.com/radio2"
864 assert parsed[1].title == "Classic Rock Radio"
865 assert parse_m3u_playlist_name(m3u_data) == "Radio Stations"
866
867
868def test_zero_duration_track_round_trips_unknown_length() -> None:
869 """Test that a zero-duration track gets an #EXTINF:-1 line that parses back to None."""
870 track = Track(
871 item_id="t1",
872 provider="builtin",
873 name="Unknown Length",
874 duration=0,
875 provider_mappings={
876 ProviderMapping(
877 item_id="http://example.com/live.mp3",
878 provider_domain="builtin",
879 provider_instance="builtin",
880 ),
881 },
882 )
883
884 m3u_data = generate_m3u("Playlist", [media_item_to_playlist_item(track)])
885
886 assert "#EXTINF:-1,Unknown Length" in m3u_data
887 parsed = parse_m3u(m3u_data)
888 assert len(parsed) == 1
889 assert parsed[0].length is None
890 assert parsed[0].title == "Unknown Length"
891
892
893# --------------------------------------------------------------------------- #
894# construct_media_item_from_playlist_item â entries without #EXTPROV #
895# --------------------------------------------------------------------------- #
896
897
898def _mass_with_builtin() -> MagicMock:
899 """Return a mock mass whose only resolvable provider is builtin."""
900
901 class DummyProvider:
902 domain = "builtin"
903 instance_id = "builtin"
904
905 mass = MagicMock()
906 mass.get_provider.side_effect = lambda ref: DummyProvider() if ref == "builtin" else None
907 return mass
908
909
910def test_construct_plain_url_gets_builtin_mapping() -> None:
911 """A bare stream URL with no #EXTPROV must still produce a usable provider mapping."""
912 item = PlaylistItem(path="http://stream.example.com/radio1")
913
914 media_item = construct_media_item_from_playlist_item(
915 item, cast("Any", _mass_with_builtin()), MediaType.RADIO
916 )
917
918 assert isinstance(media_item, Radio)
919 # an item with no mappings never reaches library_add and stays unplayable
920 assert len(media_item.provider_mappings) == 1
921 mapping = next(iter(media_item.provider_mappings))
922 assert mapping.provider_domain == "builtin"
923 assert mapping.available is True
924 # builtin's item_id *is* the stream url, so the whole url carries through
925 assert mapping.item_id == "http://stream.example.com/radio1"
926 assert media_item.item_id == "http://stream.example.com/radio1"
927
928
929@pytest.mark.parametrize(
930 "path",
931 [
932 "some/relative/file.mp3",
933 "/media/library/file.mp3",
934 # an MA-style provider URI is not a stream URL; builtin would ffprobe it and fail
935 "radiobrowser://radio/123",
936 "file://host/share/file.mp3",
937 ],
938)
939def test_construct_non_stream_url_gets_no_fallback_mapping(path: str) -> None:
940 """Only a plain stream URL falls back to builtin; anything else stays unmapped."""
941 media_item = construct_media_item_from_playlist_item(
942 PlaylistItem(path=path), cast("Any", _mass_with_builtin()), MediaType.RADIO
943 )
944
945 assert media_item is not None
946 # claiming builtin availability here would mask an entry that cannot be played
947 assert media_item.provider_mappings == set()
948
949
950def test_construct_defaults_to_requested_media_type() -> None:
951 """An entry without #EXTMA media_type honours the caller's default instead of Track."""
952 item = PlaylistItem(path="http://stream.example.com/radio1")
953 mass = cast("Any", _mass_with_builtin())
954
955 assert isinstance(
956 construct_media_item_from_playlist_item(item, mass, MediaType.RADIO),
957 Radio,
958 )
959 # the default stays Track for every existing caller
960 assert isinstance(construct_media_item_from_playlist_item(item, mass), Track)
961
962
963def test_construct_keeps_the_provider_instance() -> None:
964 """The entry's own instance is used, not whichever instance of the domain loaded first."""
965
966 class DummyProvider:
967 def __init__(self, domain: str, instance_id: str) -> None:
968 self.domain = domain
969 self.instance_id = instance_id
970
971 first = DummyProvider("radiobrowser", "radiobrowser--AAA")
972 second = DummyProvider("radiobrowser", "radiobrowser--BBB")
973 providers = {
974 # a bare domain lookup resolves to the first configured instance
975 "radiobrowser": first,
976 "radiobrowser--AAA": first,
977 "radiobrowser--BBB": second,
978 }
979 mass = MagicMock()
980 mass.get_provider.side_effect = lambda ref: providers.get(ref)
981 item = PlaylistItem(
982 path="radiobrowser://radio/station-9",
983 metadata={"media_type": MediaType.RADIO.value, "name": "Station Nine"},
984 providers=[ProviderMappingInfo("radiobrowser", "station-9", "radiobrowser--BBB")],
985 )
986
987 media_item = construct_media_item_from_playlist_item(item, cast("Any", mass), MediaType.RADIO)
988
989 assert media_item is not None
990 assert media_item.provider == "radiobrowser--BBB"
991
992
993# --------------------------------------------------------------------------- #
994# PlaylistController.tracks() â provider-driven pagination #
995# --------------------------------------------------------------------------- #
996
997
998def _make_controller() -> PlaylistController:
999 """Return a PlaylistController with a minimal mock mass."""
1000 controller = PlaylistController.__new__(PlaylistController)
1001 controller.mass = MagicMock()
1002 return controller
1003
1004
1005@pytest.mark.asyncio
1006async def test_tracks_relays_all_provider_pages() -> None:
1007 """tracks() relays every non-empty provider page until exhausted (provider decides size)."""
1008 controller = _make_controller()
1009 page0 = [MagicMock(spec=Track) for _ in range(20)]
1010 page1 = [MagicMock(spec=Track) for _ in range(15)]
1011 get_tracks = AsyncMock(side_effect=[page0, page1, []])
1012 cast("Any", controller)._get_provider_playlist_tracks = get_tracks
1013
1014 result = [t async for t in controller.tracks("abc", "some_provider")]
1015
1016 # No controller-side cap: a multi-page (e.g. dynamic) list is relayed in full.
1017 assert len(result) == 35
1018 assert get_tracks.await_count == 3
1019
1020
1021@pytest.mark.asyncio
1022async def test_tracks_stops_on_empty_page() -> None:
1023 """tracks() stops fetching once the provider yields an empty page."""
1024 controller = _make_controller()
1025 page0 = [MagicMock(spec=Track) for _ in range(7)]
1026 get_tracks = AsyncMock(side_effect=[page0, []])
1027 cast("Any", controller)._get_provider_playlist_tracks = get_tracks
1028
1029 result = [t async for t in controller.tracks("abc", "some_provider")]
1030
1031 assert len(result) == 7
1032 assert get_tracks.await_count == 2
1033
1034
1035# --------------------------------------------------------------------------- #
1036# fetch_playlist #
1037# --------------------------------------------------------------------------- #
1038
1039M3U_PLAYLIST = "#EXTM3U\n#EXTINF:-1,Test Station\nhttp://stream.example.com/aac\n"
1040PLS_PLAYLIST = (
1041 "[playlist]\n"
1042 "NumberOfEntries=1\n"
1043 "File1=http://stream.example.com/aac\n"
1044 "Title1=Test Station\n"
1045 "Length1=-1\n"
1046)
1047HLS_MEDIA_PLAYLIST = (
1048 "#EXTM3U\n"
1049 "#EXT-X-VERSION:3\n"
1050 "#EXT-X-TARGETDURATION:10\n"
1051 "#EXTINF:10.0,\n"
1052 "http://stream.example.com/segment1.aac\n"
1053)
1054VERSIONLESS_HLS_MEDIA_PLAYLIST = (
1055 "#EXTM3U\n"
1056 "#EXT-X-TARGETDURATION:10\n"
1057 '#EXT-X-KEY:METHOD=AES-128,URI="skd://test-key"\n'
1058 "#EXTINF:10,\n"
1059 "segment1.aac\n"
1060)
1061HLS_MASTER_PLAYLIST = (
1062 "#EXTM3U\n"
1063 '#EXT-X-STREAM-INF:BANDWIDTH=64000,CODECS="mp4a.40.2"\n'
1064 "http://stream.example.com/low.m3u8\n"
1065)
1066
1067
1068class _FakeContent:
1069 """Stand-in for the payload stream of an aiohttp response."""
1070
1071 def __init__(self, raw_data: bytes, chunk_size: int | None = None) -> None:
1072 self._raw_data = raw_data
1073 # like the real stream, a read hands over what has arrived so far, not the full
1074 # amount asked for
1075 self._chunk_size = chunk_size or max(len(raw_data), 1)
1076 self._pos = 0
1077
1078 async def read(self, n: int = -1) -> bytes:
1079 available = len(self._raw_data) - self._pos
1080 size = available if n < 0 else min(n, self._chunk_size, available)
1081 chunk = self._raw_data[self._pos : self._pos + size]
1082 self._pos += size
1083 return chunk
1084
1085
1086class _FakeResponse:
1087 """Stand-in for the aiohttp response of a playlist fetch."""
1088
1089 def __init__(
1090 self,
1091 raw_data: bytes,
1092 charset: str | None,
1093 status: int = 200,
1094 chunk_size: int | None = None,
1095 ) -> None:
1096 self.charset = charset
1097 self.content = _FakeContent(raw_data, chunk_size)
1098 self.status = status
1099
1100 def raise_for_status(self) -> None:
1101 if self.status >= 400:
1102 raise client_exceptions.ClientResponseError(
1103 request_info=MagicMock(),
1104 history=(),
1105 status=self.status,
1106 )
1107
1108 async def __aenter__(self) -> Self:
1109 return self
1110
1111 async def __aexit__(
1112 self,
1113 exc_type: type[BaseException] | None,
1114 exc_val: BaseException | None,
1115 exc_tb: TracebackType | None,
1116 ) -> None:
1117 return None
1118
1119
1120class _FailingRequest:
1121 """Stand-in for a playlist request that fails before a response is available."""
1122
1123 def __init__(self, error: BaseException) -> None:
1124 self._error = error
1125
1126 async def __aenter__(self) -> Self:
1127 raise self._error
1128
1129 async def __aexit__(
1130 self,
1131 exc_type: type[BaseException] | None,
1132 exc_val: BaseException | None,
1133 exc_tb: TracebackType | None,
1134 ) -> None:
1135 return None
1136
1137
1138def _mass_serving(
1139 raw_data: bytes,
1140 charset: str | None = None,
1141 status: int = 200,
1142 chunk_size: int | None = None,
1143) -> Any:
1144 """
1145 Return a mock mass whose http session serves the given playlist bytes.
1146
1147 :param raw_data: Raw response body handed to fetch_playlist.
1148 :param charset: Charset the server declares in its Content-Type header, if any.
1149 :param status: HTTP status the server answers with.
1150 :param chunk_size: Largest amount a single read hands over, if the body is chunked.
1151 """
1152 mass = MagicMock()
1153 mass.http_session.get = MagicMock(
1154 return_value=_FakeResponse(raw_data, charset, status, chunk_size)
1155 )
1156 return mass
1157
1158
1159def _mass_failing(error: BaseException) -> Any:
1160 """
1161 Return a mock mass whose http session fails the request with the given error.
1162
1163 :param error: Exception raised when the request is entered.
1164 """
1165 mass = MagicMock()
1166 mass.http_session.get = MagicMock(return_value=_FailingRequest(error))
1167 return mass
1168
1169
1170@pytest.mark.asyncio
1171async def test_fetch_playlist_timeout() -> None:
1172 """A timed out fetch is reported as invalid data instead of surfacing raw."""
1173 mass = _mass_failing(TimeoutError())
1174
1175 with pytest.raises(InvalidDataError, match="Timeout while fetching playlist"):
1176 await fetch_playlist(mass, "http://example.com/station.m3u")
1177
1178
1179@pytest.mark.asyncio
1180async def test_fetch_playlist_client_error() -> None:
1181 """A connection failure is reported as invalid data instead of surfacing raw."""
1182 mass = _mass_failing(client_exceptions.ClientConnectionError("boom"))
1183
1184 with pytest.raises(InvalidDataError, match="Error while fetching playlist"):
1185 await fetch_playlist(mass, "http://example.com/station.m3u")
1186
1187
1188@pytest.mark.asyncio
1189async def test_fetch_playlist_error_status() -> None:
1190 """
1191 An error response is rejected instead of parsed.
1192
1193 Without the status check every markup line of the error page becomes an entry.
1194 """
1195 error_page = (
1196 b"<html>\n<head><title>404 Not Found</title></head>\n"
1197 b"<body>\n<center><h1>404 Not Found</h1></center>\n</body>\n</html>\n"
1198 )
1199 mass = _mass_serving(error_page, status=404)
1200
1201 with pytest.raises(InvalidDataError, match="Error while fetching playlist"):
1202 await fetch_playlist(mass, "http://example.com/station.m3u")
1203
1204
1205@pytest.mark.asyncio
1206async def test_fetch_playlist_hls_media_playlist() -> None:
1207 """An HLS media playlist is rejected for callers that cannot handle segments."""
1208 mass = _mass_serving(HLS_MEDIA_PLAYLIST.encode())
1209
1210 with pytest.raises(IsHLSPlaylist):
1211 await fetch_playlist(mass, "http://example.com/station.m3u8")
1212
1213
1214@pytest.mark.asyncio
1215async def test_fetch_playlist_versionless_hls_media_playlist() -> None:
1216 """A version-less HLS media playlist is recognised by its required tag."""
1217 mass = _mass_serving(VERSIONLESS_HLS_MEDIA_PLAYLIST.encode())
1218
1219 with pytest.raises(IsHLSPlaylist):
1220 await fetch_playlist(mass, "http://example.com/station.m3u8")
1221
1222
1223@pytest.mark.asyncio
1224async def test_fetch_playlist_hls_media_playlist_allowed() -> None:
1225 """With raise_on_hls disabled an HLS media playlist parses like any other M3U."""
1226 mass = _mass_serving(HLS_MEDIA_PLAYLIST.encode())
1227
1228 result = await fetch_playlist(mass, "http://example.com/station.m3u8", raise_on_hls=False)
1229
1230 assert len(result) == 1
1231 assert result[0].path == "http://stream.example.com/segment1.aac"
1232
1233
1234@pytest.mark.asyncio
1235async def test_fetch_playlist_versionless_hls_media_playlist_allowed() -> None:
1236 """Disabling HLS detection still exposes a version-less playlist's segment and key."""
1237 mass = _mass_serving(VERSIONLESS_HLS_MEDIA_PLAYLIST.encode())
1238
1239 result = await fetch_playlist(mass, "http://example.com/station.m3u8", raise_on_hls=False)
1240
1241 assert len(result) == 1
1242 assert result[0].path == "segment1.aac"
1243 assert result[0].key == "skd://test-key"
1244
1245
1246@pytest.mark.asyncio
1247async def test_fetch_playlist_hls_master_playlist_always_raises() -> None:
1248 """A master playlist holds no playable entries, so it is rejected either way."""
1249 mass = _mass_serving(HLS_MASTER_PLAYLIST.encode())
1250
1251 with pytest.raises(IsHLSPlaylist):
1252 await fetch_playlist(mass, "http://example.com/station.m3u8", raise_on_hls=False)
1253
1254
1255@pytest.mark.asyncio
1256async def test_fetch_playlist_pls_by_extension(monkeypatch: pytest.MonkeyPatch) -> None:
1257 """A .pls url picks the PLS parser on the extension alone."""
1258 # every real PLS body carries the marker too, so only a marker-free body can
1259 # show which of the two conditions selected the parser
1260 parsed = [PlaylistItem(path="http://stream.example.com/aac")]
1261 parse_pls_mock = MagicMock(return_value=parsed)
1262 monkeypatch.setattr(playlists, "parse_pls", parse_pls_mock)
1263 mass = _mass_serving(M3U_PLAYLIST.encode())
1264
1265 result = await fetch_playlist(mass, "http://example.com/station.pls")
1266
1267 parse_pls_mock.assert_called_once_with(M3U_PLAYLIST)
1268 assert result == parsed
1269
1270
1271@pytest.mark.asyncio
1272async def test_fetch_playlist_pls_by_marker() -> None:
1273 """PLS content behind a url without the extension is still parsed as PLS."""
1274 mass = _mass_serving(PLS_PLAYLIST.encode())
1275
1276 result = await fetch_playlist(mass, "http://example.com/listen")
1277
1278 assert len(result) == 1
1279 assert result[0].path == "http://stream.example.com/aac"
1280 assert result[0].title == "Test Station"
1281
1282
1283@pytest.mark.asyncio
1284async def test_fetch_playlist_m3u() -> None:
1285 """Anything else is parsed as M3U."""
1286 mass = _mass_serving(M3U_PLAYLIST.encode())
1287
1288 result = await fetch_playlist(mass, "http://example.com/station.m3u")
1289
1290 assert len(result) == 1
1291 assert result[0].path == "http://stream.example.com/aac"
1292 assert result[0].title == "Test Station"
1293 assert result[0].length is None
1294
1295
1296@pytest.mark.asyncio
1297async def test_fetch_playlist_empty() -> None:
1298 """A playlist without a single entry is rejected."""
1299 mass = _mass_serving(b"#EXTM3U\n")
1300
1301 with pytest.raises(InvalidDataError, match="Empty playlist"):
1302 await fetch_playlist(mass, "http://example.com/station.m3u")
1303
1304
1305@pytest.mark.asyncio
1306async def test_fetch_playlist_unknown_charset_falls_back_to_detection() -> None:
1307 """
1308 A charset the remote server made up must not break the fetch.
1309
1310 Stations do send names Python has no codec for, which decode() answers with a
1311 LookupError that no caller on this path catches.
1312 """
1313 mass = _mass_serving(M3U_PLAYLIST.encode(), charset="utf8mb4")
1314
1315 result = await fetch_playlist(mass, "http://example.com/station.m3u")
1316
1317 assert len(result) == 1
1318 assert result[0].path == "http://stream.example.com/aac"
1319
1320
1321@pytest.mark.asyncio
1322async def test_fetch_playlist_undecodable_byte_degrades_instead_of_raising() -> None:
1323 """One bad byte costs a character, not the whole playlist."""
1324 raw_data = M3U_PLAYLIST.encode().replace(b"#EXTM3U", b"#EXTM3U\n#\xff")
1325 mass = _mass_serving(raw_data, charset="utf-8")
1326
1327 result = await fetch_playlist(mass, "http://example.com/station.m3u")
1328
1329 assert len(result) == 1
1330 assert result[0].path == "http://stream.example.com/aac"
1331
1332
1333@pytest.mark.asyncio
1334async def test_fetch_playlist_declared_charset_is_used() -> None:
1335 """A legacy charset the server declares is taken over guesswork."""
1336 # a mostly-ASCII body gives the detector too little to go on, so a station that
1337 # names its charset is the only thing keeping such a title readable
1338 raw_data = "#EXTM3U\n#EXTINF:-1,ХиÑ\nhttp://stream.example.com/aac\n".encode("cp1251")
1339 mass = _mass_serving(raw_data, charset="cp1251")
1340
1341 result = await fetch_playlist(mass, "http://example.com/station.m3u")
1342
1343 assert result[0].title == "ХиÑ"
1344
1345
1346@pytest.mark.asyncio
1347async def test_fetch_playlist_reads_a_body_that_arrives_in_chunks() -> None:
1348 """A playlist spread over several chunks is parsed whole, not just its first chunk."""
1349 mass = _mass_serving(
1350 b"#EXTM3U\n#EXTINF:-1,Station\nhttp://stream.example.com/aac\n", chunk_size=8
1351 )
1352
1353 result = await fetch_playlist(mass, "http://example.com/station.m3u")
1354
1355 assert [x.path for x in result] == ["http://stream.example.com/aac"]
1356
1357
1358@pytest.mark.asyncio
1359async def test_fetch_playlist_reads_only_the_head_of_the_body() -> None:
1360 """An oversized playlist is truncated instead of being pulled in whole."""
1361 padding = "#" + " " * (64 * 1024)
1362 mass = _mass_serving(f"#EXTM3U\n{padding}\nhttp://stream.example.com/aac\n".encode())
1363
1364 # the entry sits past the read limit, so nothing is left to parse
1365 with pytest.raises(InvalidDataError, match="Empty playlist"):
1366 await fetch_playlist(mass, "http://example.com/station.m3u")
1367