music-assistant-server

1"""
2Parity test for the summary fast path of the library list endpoints.
3
4``library_items(summary=True)`` returns slim summary variants of the media item
5models, built directly from a slim SQL query instead of hydrating the full stored
6item JSON. Every field a summary item carries must match the full item. Provider
7mappings are carried in full (so clients keep the quality badge, availability,
8provider icon and in-library state); the rest of the serialized (wire) form stays
9sparse: no null values and none of the heavy metadata fields.
10"""
11
12from __future__ import annotations
13
14from collections.abc import Iterator
15from contextlib import contextmanager
16from functools import partial
17from typing import TYPE_CHECKING, Any, cast
18
19from music_assistant_models.enums import ImageType
20from music_assistant_models.media_items import (
21    Audiobook,
22    ItemMapping,
23    MediaItemImage,
24    Playlist,
25    ProviderMapping,
26    Radio,
27    UniqueList,
28)
29from music_assistant_models.media_items.metadata import IMAGE_PROXY_ID_RESOLVER
30from music_assistant_models.translations import TRANSLATION_RESOLVER
31
32from tests.common import wait_for_sync_completion
33from tests.integration.conftest import wait_for
34
35if TYPE_CHECKING:
36    from music_assistant.controllers.music.media.base import MediaControllerBase
37    from music_assistant.mass import MusicAssistant
38
39# object-level fields a summary item may carry; compared against the full item when present
40COMPARED_FIELDS = (
41    "name",
42    "sort_name",
43    "favorite",
44    "available",
45    "version",
46    "duration",
47    "year",
48    "album_type",
49    "artist_type",
50    "owner",
51    "is_editable",
52    "is_dynamic",
53    "supported_mediatypes",
54    "publisher",
55    "total_episodes",
56    "translation_key",
57    "content_type",
58    "fully_played",
59    "resume_position_ms",
60    "disc_number",
61    "track_number",
62)
63
64
65def _assert_no_none_values(obj: Any, path: str) -> None:
66    """Recursively assert a serialized summary dict contains no null values."""
67    if isinstance(obj, dict):
68        for key, value in obj.items():
69            assert value is not None, f"unexpected null value at {path}.{key}"
70            _assert_no_none_values(value, f"{path}.{key}")
71    elif isinstance(obj, list):
72        for index, value in enumerate(obj):
73            _assert_no_none_values(value, f"{path}[{index}]")
74
75
76@contextmanager
77def _api_serialization_context(mass: MusicAssistant) -> Iterator[None]:
78    """Bind the per-request resolvers the API layer sets during serialization."""
79    token = IMAGE_PROXY_ID_RESOLVER.set(mass.metadata.compute_image_id)
80    token_loc = TRANSLATION_RESOLVER.set(partial(mass.translations.get_translation, locale="en_US"))
81    try:
82        yield
83    finally:
84        IMAGE_PROXY_ID_RESOLVER.reset(token)
85        TRANSLATION_RESOLVER.reset(token_loc)
86
87
88def _mapping_names(value: Any) -> list[tuple[str, str]]:
89    """Normalize a list of artist mappings/strings to comparable (item_id, name) pairs."""
90    result = []
91    for entry in value or []:
92        if isinstance(entry, str):
93            result.append(("", entry))
94        else:
95            result.append((str(entry.item_id), entry.name))
96    return result
97
98
99def _first_thumb(item: Any) -> MediaItemImage | None:
100    """Return the first thumb image of a media item (or None)."""
101    for image in item.metadata.images or []:
102        if image.type == ImageType.THUMB:
103            return cast("MediaItemImage", image)
104    return None
105
106
107def _sorted_provider_mappings(item_dict: dict[str, Any]) -> list[dict[str, Any]]:
108    """Return a serialized item's provider mappings, sorted for stable comparison."""
109    mappings = item_dict.get("provider_mappings") or []
110    return sorted(mappings, key=lambda m: (m["provider_instance"], m["item_id"]))
111
112
113async def _seed_playlist_and_radio(mass: MusicAssistant) -> None:
114    """Add a playlist and radio station to the library (the test provider syncs neither)."""
115    test_prov = mass.get_provider("test")
116    assert test_prov is not None
117    image = MediaItemImage(
118        type=ImageType.THUMB,
119        path="http://example.com/thumb.jpg",
120        provider=test_prov.instance_id,
121        remotely_accessible=True,
122    )
123    playlist = Playlist(
124        item_id="pl_1",
125        provider=test_prov.instance_id,
126        name="Test Playlist",
127        owner="tester",
128        is_editable=True,
129        provider_mappings={
130            ProviderMapping(
131                item_id="pl_1",
132                provider_domain="test",
133                provider_instance=test_prov.instance_id,
134                in_library=True,
135            )
136        },
137    )
138    playlist.metadata.images = UniqueList([image])
139    playlist.metadata.description = "A hand-curated test playlist."
140    await mass.music.playlists.add_item_to_library(playlist)
141    radio = Radio(
142        item_id="radio_1",
143        provider=test_prov.instance_id,
144        name="Test Radio",
145        provider_mappings={
146            ProviderMapping(
147                item_id="radio_1",
148                provider_domain="test",
149                provider_instance=test_prov.instance_id,
150                in_library=True,
151            )
152        },
153    )
154    radio.metadata.images = UniqueList([image])
155    radio.metadata.description = "A test radio station."
156    await mass.music.radio.add_item_to_library(radio)
157
158
159async def test_library_summary_items_parity(e2e_mass: MusicAssistant) -> None:
160    """Summary items must match their full counterparts on every retained field."""
161    mass = e2e_mass
162    # wait for the sync of the test provider and the follow-up genre scan
163    # so the library is fully populated
164    async with wait_for_sync_completion(mass):
165        await mass.music.start_sync()
166    await wait_for(lambda: not mass.music.active_sync_tasks)
167    await _seed_playlist_and_radio(mass)
168
169    controllers: tuple[MediaControllerBase[Any], ...] = (
170        mass.music.artists,
171        mass.music.albums,
172        mass.music.tracks,
173        mass.music.playlists,
174        mass.music.radio,
175        mass.music.audiobooks,
176        mass.music.podcasts,
177        mass.music.genres,
178    )
179    for ctrl in controllers:
180        full_items = await ctrl.library_items(order_by="name")
181        summary_items = await ctrl.library_items(order_by="name", summary=True)
182        assert len(full_items) > 0, f"no library items for {ctrl.media_type.value}"
183        assert [x.item_id for x in summary_items] == [x.item_id for x in full_items]
184
185        for full, item in zip(full_items, summary_items, strict=True):
186            # same model type (summary subclass), same identity
187            assert isinstance(item, ctrl.item_cls)
188            assert type(item) is ctrl.summary_item_cls
189            assert item.provider == "library"
190            assert item.uri == full.uri
191            for field_name in COMPARED_FIELDS:
192                if not hasattr(item, field_name):
193                    continue
194                assert getattr(item, field_name) == getattr(full, field_name), (
195                    f"{ctrl.media_type.value} {item.item_id}: field {field_name} differs"
196                )
197            # artist/album/author mappings resolve to the same library items
198            if hasattr(item, "artists"):
199                assert _mapping_names(item.artists) == _mapping_names(full.artists)
200            if hasattr(item, "album") and full.album:
201                assert isinstance(item.album, ItemMapping)
202                assert item.album.item_id == full.album.item_id
203                assert item.album.name == full.album.name
204                assert item.album.year == full.album.year
205            if isinstance(item, Audiobook):
206                assert isinstance(full, Audiobook)
207                assert _mapping_names(item.authors) == _mapping_names(full.authors)
208                assert _mapping_names(item.narrators) == _mapping_names(full.narrators)
209            # the (single) summary image is the same image the full item shows first
210            full_thumb = _first_thumb(full)
211            summary_thumb = _first_thumb(item)
212            if full_thumb:
213                assert summary_thumb is not None
214                assert summary_thumb.path == full_thumb.path
215                assert summary_thumb.provider == full_thumb.provider
216
217            # wire format: provider mappings are carried in full and must match the
218            # full item exactly (the frontend reads audio_format for the quality badge,
219            # availability, the provider icon and in-library state from here); the rest
220            # of the wire form stays sparse (no nulls) and free of heavy metadata, and
221            # the image dicts carry the same resolved proxy_id / localized names; bind
222            # the same resolvers the API layer sets while serializing a response
223            with _api_serialization_context(mass):
224                full_dict = full.to_dict()
225                summary_dict = item.to_dict()
226            assert _sorted_provider_mappings(summary_dict) == _sorted_provider_mappings(full_dict)
227            wire = {k: v for k, v in summary_dict.items() if k != "provider_mappings"}
228            _assert_no_none_values(wire, ctrl.media_type.value)
229            assert summary_dict["name"] == full_dict["name"]
230            # summary metadata carries only the thumb image plus the few descriptive
231            # fields the list rows render: the explicit flag and release date (tracks) and
232            # the description (radio/playlist). A description may also be injected by the
233            # translation resolver for localizable items; either way both modes must carry
234            # the same value.
235            assert set(summary_dict["metadata"]) <= {
236                "images",
237                "explicit",
238                "release_date",
239                "description",
240            }
241            for meta_field in ("description", "release_date"):
242                if meta_field in summary_dict["metadata"]:
243                    assert summary_dict["metadata"][meta_field] == full_dict["metadata"][meta_field]
244            if full_thumb:
245                assert summary_dict["metadata"]["images"][0] == full_dict["metadata"]["images"][0]
246                assert summary_dict["metadata"]["images"][0]["proxy_id"]
247
248
249async def test_library_summary_items_filters(e2e_mass: MusicAssistant) -> None:
250    """The summary fast path composes with the regular list filters and paging."""
251    mass = e2e_mass
252    async with wait_for_sync_completion(mass):
253        await mass.music.start_sync()
254    await wait_for(lambda: not mass.music.active_sync_tasks)
255
256    full_items = await mass.music.tracks.library_items(order_by="name")
257    assert len(full_items) > 2
258    # paging
259    page = await mass.music.tracks.library_items(order_by="name", limit=2, offset=1, summary=True)
260    assert [x.item_id for x in page] == [x.item_id for x in full_items[1:3]]
261    # search
262    needle = full_items[0].name
263    full_search = await mass.music.tracks.library_items(search=needle)
264    summary_search = await mass.music.tracks.library_items(search=needle, summary=True)
265    assert sorted(x.item_id for x in summary_search) == sorted(x.item_id for x in full_search)
266    # favorite filter
267    await mass.music.tracks.set_favorite(full_items[0].item_id, True)
268    favorites = await mass.music.tracks.library_items(favorite=True, summary=True)
269    assert [x.item_id for x in favorites] == [full_items[0].item_id]
270    assert favorites[0].favorite is True
271