/
/
/
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