music-assistant-server

1"""Test we can parse Open Subsonic models into Music Assistant models."""
2
3import logging
4import pathlib
5
6import aiofiles
7import pytest
8from libopensonic.media import (
9    AlbumID3,
10    AlbumInfo,
11    ArtistID3,
12    ArtistInfo,
13    Child,
14    Playlist,
15    PodcastChannel,
16    PodcastEpisode,
17)
18from syrupy.assertion import SnapshotAssertion
19
20from music_assistant.providers.opensubsonic.parsers import (
21    parse_album,
22    parse_artist,
23    parse_epsiode,
24    parse_playlist,
25    parse_podcast,
26    parse_track,
27)
28
29FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures"
30ARTIST_FIXTURES = list(FIXTURES_DIR.glob("artists/*.artist.json"))
31ALBUM_FIXTURES = list(FIXTURES_DIR.glob("albums/*.album.json"))
32PLAYLIST_FIXTURES = list(FIXTURES_DIR.glob("playlists/*.playlist.json"))
33PODCAST_FIXTURES = list(FIXTURES_DIR.glob("podcasts/*.podcast.json"))
34EPISODE_FIXTURES = list(FIXTURES_DIR.glob("episodes/*.episode.json"))
35TRACK_FIXTURES = list(FIXTURES_DIR.glob("tracks/*.track.json"))
36
37_LOGGER = logging.getLogger(__name__)
38
39
40@pytest.mark.parametrize("example", ARTIST_FIXTURES, ids=lambda val: str(val.stem))
41async def test_parse_artists(example: pathlib.Path, snapshot: SnapshotAssertion) -> None:
42    """Test we can parse artists."""
43    async with aiofiles.open(example) as fp:
44        artist = ArtistID3.from_json(await fp.read())
45
46    parsed = parse_artist("xx-instance-id-xx", artist).to_dict()
47    # sort external Ids to ensure they are always in the same order for snapshot testing
48    parsed["external_ids"].sort()
49    assert snapshot == parsed
50
51    # Find the corresponding info file
52    example_info = example.with_suffix("").with_suffix(".info.json")
53    async with aiofiles.open(example_info) as fp:
54        artist_info = ArtistInfo.from_json(await fp.read())
55
56    parsed = parse_artist("xx-instance-id-xx", artist, artist_info).to_dict()
57    # sort external Ids to ensure they are always in the same order for snapshot testing
58    parsed["external_ids"].sort()
59    assert snapshot == parsed
60
61
62@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: str(val.stem))
63async def test_parse_albums(example: pathlib.Path, snapshot: SnapshotAssertion) -> None:
64    """Test we can parse albums."""
65    async with aiofiles.open(example) as fp:
66        album = AlbumID3.from_json(await fp.read())
67
68    parsed = parse_album(_LOGGER, "xx-instance-id-xx", album).to_dict()
69    # sort external Ids and genres to ensure they are always in the same order for snapshot testing
70    parsed["external_ids"].sort()
71    parsed["metadata"]["genres"].sort()
72    assert snapshot == parsed
73
74    # Find the corresponding info file
75    example_info = example.with_suffix("").with_suffix(".info.json")
76    async with aiofiles.open(example_info) as fp:
77        album_info = AlbumInfo.from_json(await fp.read())
78
79    parsed = parse_album(_LOGGER, "xx-instance-id-xx", album, album_info).to_dict()
80    # sort external Ids and genres to ensure they are always in the same order for snapshot testing
81    parsed["external_ids"].sort()
82    parsed["metadata"]["genres"].sort()
83    assert snapshot == parsed
84
85
86@pytest.mark.parametrize("example", PLAYLIST_FIXTURES, ids=lambda val: str(val.stem))
87async def test_parse_playlist(example: pathlib.Path, snapshot: SnapshotAssertion) -> None:
88    """Test we can parse Playlists."""
89    async with aiofiles.open(example) as fp:
90        playlist = Playlist.from_json(await fp.read())
91
92    parsed = parse_playlist("xx-instance-id-xx", playlist).to_dict()
93    # sort external Ids to ensure they are always in the same order for snapshot testing
94    parsed["external_ids"].sort()
95    assert snapshot == parsed
96
97
98@pytest.mark.parametrize("example", PODCAST_FIXTURES, ids=lambda val: str(val.stem))
99async def test_parse_podcast(example: pathlib.Path, snapshot: SnapshotAssertion) -> None:
100    """Test we can parse Podcasts."""
101    async with aiofiles.open(example) as fp:
102        podcast = PodcastChannel.from_json(await fp.read())
103
104    parsed = parse_podcast("xx-instance-id-xx", podcast).to_dict()
105    # sort external Ids to ensure they are always in the same order for snapshot testing
106    parsed["external_ids"].sort()
107    assert snapshot == parsed
108
109
110@pytest.mark.parametrize("example", EPISODE_FIXTURES, ids=lambda val: str(val.stem))
111async def test_parse_episode(example: pathlib.Path, snapshot: SnapshotAssertion) -> None:
112    """Test we can parse Podcast Episodes."""
113    async with aiofiles.open(example) as fp:
114        episode = PodcastEpisode.from_json(await fp.read())
115
116    example_channel = example.with_suffix("").with_suffix(".podcast.json")
117    async with aiofiles.open(example_channel) as fp:
118        channel = PodcastChannel.from_json(await fp.read())
119
120    parsed = parse_epsiode("xx-instance-id-xx", episode, channel).to_dict()
121    # sort external Ids to ensure they are always in the same order for snapshot testing
122    parsed["external_ids"].sort()
123    assert snapshot == parsed
124
125
126@pytest.mark.parametrize("example", TRACK_FIXTURES, ids=lambda val: str(val.stem))
127async def test_parse_track(example: pathlib.Path, snapshot: SnapshotAssertion) -> None:
128    """Test we can parse Tracks."""
129    async with aiofiles.open(example) as fp:
130        song = Child.from_json(await fp.read())
131
132    parsed = parse_track(_LOGGER, "xx-instance-id-xx", song).to_dict()
133    # sort external Ids, genres, and performers to ensure they are always in the same
134    # order for snapshot testing
135    parsed["external_ids"].sort()
136    parsed["metadata"]["genres"].sort()
137    parsed["metadata"]["performers"].sort()
138    assert snapshot == parsed
139
140    example_album = example.with_suffix("").with_suffix(".album.json")
141    async with aiofiles.open(example_album) as fp:
142        album = AlbumID3.from_json(await fp.read())
143
144    parsed = parse_track(_LOGGER, "xx-instance-id-xx", song, album).to_dict()
145    # sort external Ids, genres, and performers to ensure they are always in the same
146    # order for snapshot testing
147    parsed["external_ids"].sort()
148    parsed["metadata"]["genres"].sort()
149    parsed["metadata"]["performers"].sort()
150    assert snapshot == parsed
151