/
/
/
1"""Test we can parse Jellyfin models into Music Assistant models."""
2
3import logging
4import pathlib
5from collections.abc import AsyncGenerator
6from typing import TYPE_CHECKING, Any
7
8import aiofiles
9import aiohttp
10import pytest
11from aiojellyfin import Artist, Connection
12from aiojellyfin.session import SessionConfiguration
13from mashumaro.codecs.json import JSONDecoder
14from music_assistant_models.enums import ContentType
15
16from music_assistant.providers.jellyfin.const import (
17 ITEM_KEY_CONTAINER,
18 ITEM_KEY_MEDIA_CHANNELS,
19 ITEM_KEY_MEDIA_CODEC,
20 ITEM_KEY_MEDIA_SOURCES,
21 ITEM_KEY_MEDIA_STREAM_TYPE,
22 ITEM_KEY_MEDIA_STREAMS,
23)
24from music_assistant.providers.jellyfin.parsers import (
25 audio_format,
26 parse_album,
27 parse_artist,
28 parse_track,
29)
30
31if TYPE_CHECKING:
32 from syrupy.assertion import SnapshotAssertion
33
34FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures"
35ARTIST_FIXTURES = list(FIXTURES_DIR.glob("artists/*.json"))
36ALBUM_FIXTURES = list(FIXTURES_DIR.glob("albums/*.json"))
37TRACK_FIXTURES = list(FIXTURES_DIR.glob("tracks/*.json"))
38
39ARTIST_DECODER = JSONDecoder(Artist)
40
41_LOGGER = logging.getLogger(__name__)
42
43
44@pytest.fixture
45async def connection() -> AsyncGenerator[Connection]:
46 """Spin up a dummy connection."""
47 async with aiohttp.ClientSession() as session:
48 session_config = SessionConfiguration(
49 session=session,
50 url="http://localhost:1234",
51 app_name="X",
52 app_version="0.0.0",
53 device_id="X",
54 device_name="localhost",
55 )
56 yield Connection(session_config, "USER_ID", "ACCESS_TOKEN")
57
58
59@pytest.mark.parametrize("example", ARTIST_FIXTURES, ids=lambda val: str(val.stem))
60async def test_parse_artists(
61 example: pathlib.Path, connection: Connection, snapshot: SnapshotAssertion
62) -> None:
63 """Test we can parse artists."""
64 async with aiofiles.open(example, encoding="utf-8") as fp:
65 raw_data = ARTIST_DECODER.decode(await fp.read())
66 parsed = parse_artist(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
67 # sort external Ids to ensure they are always in the same order for snapshot testing
68 parsed["external_ids"].sort()
69 assert snapshot == parsed
70
71
72@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: str(val.stem))
73async def test_parse_albums(
74 example: pathlib.Path, connection: Connection, snapshot: SnapshotAssertion
75) -> None:
76 """Test we can parse albums."""
77 async with aiofiles.open(example, encoding="utf-8") as fp:
78 raw_data = ARTIST_DECODER.decode(await fp.read())
79 parsed = parse_album(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
80 # sort external Ids to ensure they are always in the same order for snapshot testing
81 parsed["external_ids"].sort()
82 assert snapshot == parsed
83
84
85@pytest.mark.parametrize("example", TRACK_FIXTURES, ids=lambda val: str(val.stem))
86async def test_parse_tracks(
87 example: pathlib.Path, connection: Connection, snapshot: SnapshotAssertion
88) -> None:
89 """Test we can parse tracks."""
90 async with aiofiles.open(example, encoding="utf-8") as fp:
91 raw_data = ARTIST_DECODER.decode(await fp.read())
92 parsed = parse_track(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
93 # sort external Ids to ensure they are always in the same order for snapshot testing
94 parsed["external_ids"]
95 assert snapshot == parsed
96
97
98def test_audio_format_empty_mediastreams() -> None:
99 """Test audio_format handles empty MediaStreams array."""
100 # Track with empty MediaStreams
101 track: dict[str, Any] = {
102 ITEM_KEY_MEDIA_STREAMS: [],
103 }
104 result = audio_format(track) # type: ignore[arg-type]
105
106 # Verify no exception is raised and result has expected attributes
107 assert result is not None
108 assert hasattr(result, "content_type")
109
110
111def test_audio_format_missing_channels() -> None:
112 """Test audio_format applies default when Channels field is missing."""
113 # Track with MediaStreams but missing Channels
114 track: dict[str, Any] = {
115 ITEM_KEY_MEDIA_SOURCES: [{ITEM_KEY_CONTAINER: "mp3"}],
116 ITEM_KEY_MEDIA_STREAMS: [
117 {
118 ITEM_KEY_MEDIA_STREAM_TYPE: "Audio",
119 ITEM_KEY_MEDIA_CODEC: "mp3",
120 "SampleRate": 48000,
121 "BitDepth": 16,
122 "BitRate": 320000,
123 }
124 ],
125 }
126 result = audio_format(track) # type: ignore[arg-type]
127
128 # Verify defaults are applied correctly
129 assert result is not None
130 assert result.channels == 2 # Default stereo
131 assert result.sample_rate == 48000
132 assert result.bit_depth == 16
133 assert result.bit_rate == 320 # AudioFormat converts bps to kbps automatically
134
135
136def test_audio_format_wav_container_not_treated_as_raw_pcm() -> None:
137 """A WAV/PCM source must report the container so ffmpeg does not read it as raw PCM."""
138 # content_type must be the container (wav), not the codec (pcm_s24le): a PCM content_type
139 # makes the ffmpeg pipeline read the stream as headerless raw PCM, so the WAV header and
140 # embedded cover art get decoded as samples -> white noise. The embedded cover art is also
141 # why the audio stream is not first: the parser must pick the stream by Type, not index 0.
142 track: dict[str, Any] = {
143 ITEM_KEY_MEDIA_SOURCES: [{ITEM_KEY_CONTAINER: "wav"}],
144 ITEM_KEY_MEDIA_STREAMS: [
145 {ITEM_KEY_MEDIA_STREAM_TYPE: "Video", ITEM_KEY_MEDIA_CODEC: "mjpeg"},
146 {
147 ITEM_KEY_MEDIA_STREAM_TYPE: "Audio",
148 ITEM_KEY_MEDIA_CODEC: "pcm_s24le",
149 ITEM_KEY_MEDIA_CHANNELS: 2,
150 "SampleRate": 48000,
151 "BitDepth": 24,
152 },
153 ],
154 }
155 result = audio_format(track) # type: ignore[arg-type]
156
157 assert result.content_type == ContentType.WAV
158 assert not result.content_type.is_pcm()
159 assert result.codec_type == ContentType.PCM_S24LE
160 assert result.sample_rate == 48000
161 assert result.bit_depth == 24
162 assert result.channels == 2
163