music-assistant-server

1"""Test we can parse Jellyfin models into Music Assistant models."""
2
3import logging
4import pathlib
5from collections.abc import AsyncGenerator
6from typing import 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 syrupy.assertion import SnapshotAssertion
15
16from music_assistant.providers.jellyfin.const import (
17    ITEM_KEY_MEDIA_CODEC,
18    ITEM_KEY_MEDIA_STREAMS,
19)
20from music_assistant.providers.jellyfin.parsers import (
21    audio_format,
22    parse_album,
23    parse_artist,
24    parse_track,
25)
26
27FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures"
28ARTIST_FIXTURES = list(FIXTURES_DIR.glob("artists/*.json"))
29ALBUM_FIXTURES = list(FIXTURES_DIR.glob("albums/*.json"))
30TRACK_FIXTURES = list(FIXTURES_DIR.glob("tracks/*.json"))
31
32ARTIST_DECODER = JSONDecoder(Artist)
33
34_LOGGER = logging.getLogger(__name__)
35
36
37@pytest.fixture
38async def connection() -> AsyncGenerator[Connection, None]:
39    """Spin up a dummy connection."""
40    async with aiohttp.ClientSession() as session:
41        session_config = SessionConfiguration(
42            session=session,
43            url="http://localhost:1234",
44            app_name="X",
45            app_version="0.0.0",
46            device_id="X",
47            device_name="localhost",
48        )
49        yield Connection(session_config, "USER_ID", "ACCESS_TOKEN")
50
51
52@pytest.mark.parametrize("example", ARTIST_FIXTURES, ids=lambda val: str(val.stem))
53async def test_parse_artists(
54    example: pathlib.Path, connection: Connection, snapshot: SnapshotAssertion
55) -> None:
56    """Test we can parse artists."""
57    async with aiofiles.open(example) as fp:
58        raw_data = ARTIST_DECODER.decode(await fp.read())
59    parsed = parse_artist(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
60    # sort external Ids to ensure they are always in the same order for snapshot testing
61    parsed["external_ids"].sort()
62    assert snapshot == parsed
63
64
65@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: str(val.stem))
66async def test_parse_albums(
67    example: pathlib.Path, connection: Connection, snapshot: SnapshotAssertion
68) -> None:
69    """Test we can parse albums."""
70    async with aiofiles.open(example) as fp:
71        raw_data = ARTIST_DECODER.decode(await fp.read())
72    parsed = parse_album(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
73    # sort external Ids to ensure they are always in the same order for snapshot testing
74    parsed["external_ids"].sort()
75    assert snapshot == parsed
76
77
78@pytest.mark.parametrize("example", TRACK_FIXTURES, ids=lambda val: str(val.stem))
79async def test_parse_tracks(
80    example: pathlib.Path, connection: Connection, snapshot: SnapshotAssertion
81) -> None:
82    """Test we can parse tracks."""
83    async with aiofiles.open(example) as fp:
84        raw_data = ARTIST_DECODER.decode(await fp.read())
85    parsed = parse_track(_LOGGER, "xx-instance-id-xx", connection, raw_data).to_dict()
86    # sort external Ids to ensure they are always in the same order for snapshot testing
87    parsed["external_ids"]
88    assert snapshot == parsed
89
90
91def test_audio_format_empty_mediastreams() -> None:
92    """Test audio_format handles empty MediaStreams array."""
93    # Track with empty MediaStreams
94    track: dict[str, Any] = {
95        ITEM_KEY_MEDIA_STREAMS: [],
96    }
97    result = audio_format(track)  # type: ignore[arg-type]
98
99    # Verify no exception is raised and result has expected attributes
100    assert result is not None
101    assert hasattr(result, "content_type")
102
103
104def test_audio_format_missing_channels() -> None:
105    """Test audio_format applies default when Channels field is missing."""
106    # Track with MediaStreams but missing Channels
107    track: dict[str, Any] = {
108        ITEM_KEY_MEDIA_STREAMS: [
109            {
110                ITEM_KEY_MEDIA_CODEC: "mp3",
111                "SampleRate": 48000,
112                "BitDepth": 16,
113                "BitRate": 320000,
114            }
115        ],
116    }
117    result = audio_format(track)  # type: ignore[arg-type]
118
119    # Verify defaults are applied correctly
120    assert result is not None
121    assert result.channels == 2  # Default stereo
122    assert result.sample_rate == 48000
123    assert result.bit_depth == 16
124    assert result.bit_rate == 320  # AudioFormat converts bps to kbps automatically
125