music-assistant-server

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