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)
1003HLS_MASTER_PLAYLIST = (
1004    "#EXTM3U\n"
1005    '#EXT-X-STREAM-INF:BANDWIDTH=64000,CODECS="mp4a.40.2"\n'
1006    "http://stream.example.com/low.m3u8\n"
1007)
1008
1009
1010class _FakeContent:
1011    """Stand-in for the payload stream of an aiohttp response."""
1012
1013    def __init__(self, raw_data: bytes) -> None:
1014        self._raw_data = raw_data
1015
1016    async def read(self, n: int = -1) -> bytes:
1017        return self._raw_data if n < 0 else self._raw_data[:n]
1018
1019
1020class _FakeResponse:
1021    """Stand-in for the aiohttp response of a playlist fetch."""
1022
1023    def __init__(self, raw_data: bytes, charset: str | None) -> None:
1024        self.charset = charset
1025        self.content = _FakeContent(raw_data)
1026
1027    async def __aenter__(self) -> Self:
1028        return self
1029
1030    async def __aexit__(
1031        self,
1032        exc_type: type[BaseException] | None,
1033        exc_val: BaseException | None,
1034        exc_tb: TracebackType | None,
1035    ) -> None:
1036        return None
1037
1038
1039class _FailingRequest:
1040    """Stand-in for a playlist request that fails before a response is available."""
1041
1042    def __init__(self, error: BaseException) -> None:
1043        self._error = error
1044
1045    async def __aenter__(self) -> Self:
1046        raise self._error
1047
1048    async def __aexit__(
1049        self,
1050        exc_type: type[BaseException] | None,
1051        exc_val: BaseException | None,
1052        exc_tb: TracebackType | None,
1053    ) -> None:
1054        return None
1055
1056
1057def _mass_serving(raw_data: bytes, charset: str | None = None) -> Any:
1058    """
1059    Return a mock mass whose http session serves the given playlist bytes.
1060
1061    :param raw_data: Raw response body handed to fetch_playlist.
1062    :param charset: Charset the server declares in its Content-Type header, if any.
1063    """
1064    mass = MagicMock()
1065    mass.http_session.get = MagicMock(return_value=_FakeResponse(raw_data, charset))
1066    return mass
1067
1068
1069def _mass_failing(error: BaseException) -> Any:
1070    """
1071    Return a mock mass whose http session fails the request with the given error.
1072
1073    :param error: Exception raised when the request is entered.
1074    """
1075    mass = MagicMock()
1076    mass.http_session.get = MagicMock(return_value=_FailingRequest(error))
1077    return mass
1078
1079
1080@pytest.mark.asyncio
1081async def test_fetch_playlist_timeout() -> None:
1082    """A timed out fetch is reported as invalid data instead of surfacing raw."""
1083    mass = _mass_failing(TimeoutError())
1084
1085    with pytest.raises(InvalidDataError, match="Timeout while fetching playlist"):
1086        await fetch_playlist(mass, "http://example.com/station.m3u")
1087
1088
1089@pytest.mark.asyncio
1090async def test_fetch_playlist_client_error() -> None:
1091    """A connection failure is reported as invalid data instead of surfacing raw."""
1092    mass = _mass_failing(client_exceptions.ClientConnectionError("boom"))
1093
1094    with pytest.raises(InvalidDataError, match="Error while fetching playlist"):
1095        await fetch_playlist(mass, "http://example.com/station.m3u")
1096
1097
1098@pytest.mark.asyncio
1099async def test_fetch_playlist_hls_media_playlist() -> None:
1100    """An HLS media playlist is rejected for callers that cannot handle segments."""
1101    mass = _mass_serving(HLS_MEDIA_PLAYLIST.encode())
1102
1103    with pytest.raises(IsHLSPlaylist):
1104        await fetch_playlist(mass, "http://example.com/station.m3u8")
1105
1106
1107@pytest.mark.asyncio
1108async def test_fetch_playlist_hls_media_playlist_allowed() -> None:
1109    """With raise_on_hls disabled an HLS media playlist parses like any other M3U."""
1110    mass = _mass_serving(HLS_MEDIA_PLAYLIST.encode())
1111
1112    result = await fetch_playlist(mass, "http://example.com/station.m3u8", raise_on_hls=False)
1113
1114    assert len(result) == 1
1115    assert result[0].path == "http://stream.example.com/segment1.aac"
1116
1117
1118@pytest.mark.asyncio
1119async def test_fetch_playlist_hls_master_playlist_always_raises() -> None:
1120    """A master playlist holds no playable entries, so it is rejected either way."""
1121    mass = _mass_serving(HLS_MASTER_PLAYLIST.encode())
1122
1123    with pytest.raises(IsHLSPlaylist):
1124        await fetch_playlist(mass, "http://example.com/station.m3u8", raise_on_hls=False)
1125
1126
1127@pytest.mark.asyncio
1128async def test_fetch_playlist_pls_by_extension(monkeypatch: pytest.MonkeyPatch) -> None:
1129    """A .pls url picks the PLS parser on the extension alone."""
1130    # every real PLS body carries the marker too, so only a marker-free body can
1131    # show which of the two conditions selected the parser
1132    parsed = [PlaylistItem(path="http://stream.example.com/aac")]
1133    parse_pls_mock = MagicMock(return_value=parsed)
1134    monkeypatch.setattr(playlists, "parse_pls", parse_pls_mock)
1135    mass = _mass_serving(M3U_PLAYLIST.encode())
1136
1137    result = await fetch_playlist(mass, "http://example.com/station.pls")
1138
1139    parse_pls_mock.assert_called_once_with(M3U_PLAYLIST)
1140    assert result == parsed
1141
1142
1143@pytest.mark.asyncio
1144async def test_fetch_playlist_pls_by_marker() -> None:
1145    """PLS content behind a url without the extension is still parsed as PLS."""
1146    mass = _mass_serving(PLS_PLAYLIST.encode())
1147
1148    result = await fetch_playlist(mass, "http://example.com/listen")
1149
1150    assert len(result) == 1
1151    assert result[0].path == "http://stream.example.com/aac"
1152    assert result[0].title == "Test Station"
1153
1154
1155@pytest.mark.asyncio
1156async def test_fetch_playlist_m3u() -> None:
1157    """Anything else is parsed as M3U."""
1158    mass = _mass_serving(M3U_PLAYLIST.encode())
1159
1160    result = await fetch_playlist(mass, "http://example.com/station.m3u")
1161
1162    assert len(result) == 1
1163    assert result[0].path == "http://stream.example.com/aac"
1164    assert result[0].title == "Test Station"
1165    assert result[0].length is None
1166
1167
1168@pytest.mark.asyncio
1169async def test_fetch_playlist_empty() -> None:
1170    """A playlist without a single entry is rejected."""
1171    mass = _mass_serving(b"#EXTM3U\n")
1172
1173    with pytest.raises(InvalidDataError, match="Empty playlist"):
1174        await fetch_playlist(mass, "http://example.com/station.m3u")
1175
1176
1177@pytest.mark.asyncio
1178async def test_fetch_playlist_unknown_charset_falls_back_to_detection() -> None:
1179    """
1180    A charset the remote server made up must not break the fetch.
1181
1182    Stations do send names Python has no codec for, which decode() answers with a
1183    LookupError that no caller on this path catches.
1184    """
1185    mass = _mass_serving(M3U_PLAYLIST.encode(), charset="utf8mb4")
1186
1187    result = await fetch_playlist(mass, "http://example.com/station.m3u")
1188
1189    assert len(result) == 1
1190    assert result[0].path == "http://stream.example.com/aac"
1191
1192
1193@pytest.mark.asyncio
1194async def test_fetch_playlist_undecodable_byte_degrades_instead_of_raising() -> None:
1195    """One bad byte costs a character, not the whole playlist."""
1196    raw_data = M3U_PLAYLIST.encode().replace(b"#EXTM3U", b"#EXTM3U\n#\xff")
1197    mass = _mass_serving(raw_data, charset="utf-8")
1198
1199    result = await fetch_playlist(mass, "http://example.com/station.m3u")
1200
1201    assert len(result) == 1
1202    assert result[0].path == "http://stream.example.com/aac"
1203
1204
1205@pytest.mark.asyncio
1206async def test_fetch_playlist_declared_charset_is_used() -> None:
1207    """A legacy charset the server declares is taken over guesswork."""
1208    # a mostly-ASCII body gives the detector too little to go on, so a station that
1209    # names its charset is the only thing keeping such a title readable
1210    raw_data = "#EXTM3U\n#EXTINF:-1,Хит\nhttp://stream.example.com/aac\n".encode("cp1251")
1211    mass = _mass_serving(raw_data, charset="cp1251")
1212
1213    result = await fetch_playlist(mass, "http://example.com/station.m3u")
1214
1215    assert result[0].title == "Хит"
1216
1217
1218@pytest.mark.asyncio
1219async def test_fetch_playlist_reads_only_the_head_of_the_body() -> None:
1220    """An oversized playlist is truncated instead of being pulled in whole."""
1221    padding = "#" + " " * (64 * 1024)
1222    mass = _mass_serving(f"#EXTM3U\n{padding}\nhttp://stream.example.com/aac\n".encode())
1223
1224    # the entry sits past the read limit, so nothing is left to parse
1225    with pytest.raises(InvalidDataError, match="Empty playlist"):
1226        await fetch_playlist(mass, "http://example.com/station.m3u")
1227