/
/
/
1"""
2Tests for the library listing query builder (EXISTS-based provider filtering).
3
4The provider/in-library filters in ``get_library_items_by_query`` are applied as
5correlated EXISTS subqueries (instead of a JOIN + GROUP BY) so SQLite can stream
6results straight from the sort index. These tests verify that:
7
8* the new queries return the exact same items (and order) as the legacy
9 JOIN + GROUP BY queries for a matrix of filter combinations
10* the generated listing queries stream from the sort index without a
11 temporary b-tree for sorting
12"""
13
14from __future__ import annotations
15
16import json
17from typing import Any
18from unittest.mock import patch
19from uuid import uuid4
20
21import pytest
22from music_assistant_models.enums import AlbumType, ArtistType, MediaType
23from music_assistant_models.helpers import create_safe_string
24from music_assistant_models.media_items import (
25 Album,
26 Artist,
27 Audiobook,
28 MediaCollection,
29 MediaItemCollection,
30 MediaItemMetadata,
31 ProviderMapping,
32 Track,
33)
34from music_assistant_models.unique_list import UniqueList
35
36from music_assistant.constants import (
37 DB_TABLE_ALBUM_TRACKS,
38 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
39 DB_TABLE_PLAYLOG,
40 DB_TABLE_PROVIDER_MAPPINGS,
41)
42from music_assistant.controllers.music.media.base import MediaControllerBase
43from music_assistant.mass import MusicAssistant
44
45pytestmark = pytest.mark.asyncio
46
47
48@pytest.fixture(scope="module")
49async def seeded_mass(
50 music_mass_module: MusicAssistant,
51) -> MusicAssistant:
52 """Return a module-scoped database-only instance with a seeded library."""
53 await _seed_library(music_mass_module)
54 return music_mass_module
55
56
57def _mapping(provider_instance: str = "prov_a_inst") -> ProviderMapping:
58 """Create a provider mapping with a unique provider_item_id."""
59 return ProviderMapping(
60 item_id=uuid4().hex,
61 provider_domain=provider_instance.removesuffix("_inst"),
62 provider_instance=provider_instance,
63 in_library=True,
64 )
65
66
67async def _seed_library(mass: MusicAssistant) -> None:
68 """Seed artists, albums and tracks covering all listing edge cases."""
69 artists: list[Artist] = []
70 for idx in range(1, 5):
71 artist = Artist(
72 item_id="0",
73 provider="library",
74 name=f"Artist {idx:02d}",
75 provider_mappings={_mapping()},
76 favorite=idx % 2 == 0,
77 )
78 artists.append(await mass.music.artists.add_item_to_library(artist))
79
80 albums: list[Album] = []
81 for idx in range(1, 6):
82 album = Album(
83 item_id="0",
84 provider="library",
85 name=f"Album {idx:02d}",
86 album_type=AlbumType.ALBUM,
87 provider_mappings={
88 # some albums have multiple provider mappings (fanout in legacy JOIN)
89 _mapping(),
90 *([_mapping("prov_b_inst")] if idx % 2 == 0 else []),
91 },
92 artists=UniqueList([artists[idx % len(artists)]]),
93 favorite=idx % 3 == 0,
94 )
95 albums.append(await mass.music.albums.add_item_to_library(album))
96
97 for idx in range(1, 21):
98 track = Track(
99 item_id="0",
100 provider="library",
101 name=f"Track {idx:02d}",
102 provider_mappings={
103 _mapping(),
104 *([_mapping("prov_b_inst")] if idx % 4 == 0 else []),
105 },
106 artists=UniqueList([artists[idx % len(artists)]]),
107 album=albums[idx % len(albums)],
108 disc_number=1,
109 track_number=idx,
110 favorite=idx % 5 == 0,
111 )
112 db_track = await mass.music.tracks.add_item_to_library(track)
113 # track 3 appears on two albums (fanout in legacy album_tracks JOIN)
114 if idx == 3:
115 await mass.music.tracks._set_track_album(
116 db_id=int(db_track.item_id),
117 album=albums[4],
118 disc_number=2,
119 track_number=99,
120 )
121
122 # one track only has a mapping with in_library=0 (hidden from library listings)
123 hidden = Track(
124 item_id="0",
125 provider="library",
126 name="Track hidden",
127 provider_mappings={_mapping()},
128 artists=UniqueList([artists[0]]),
129 favorite=True,
130 )
131 db_hidden = await mass.music.tracks.add_item_to_library(hidden)
132 await mass.music.database.execute(
133 f"UPDATE {DB_TABLE_PROVIDER_MAPPINGS} SET in_library = 0 "
134 "WHERE item_id = :item_id AND media_type = 'track'",
135 {"item_id": int(db_hidden.item_id)},
136 )
137
138 # genre mapping for some tracks
139 for track_db_id in (1, 2, 3):
140 await mass.music.database.insert(
141 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
142 {"genre_id": 1, "media_id": track_db_id, "media_type": "track"},
143 )
144 # mark some items as played
145 await mass.music.database.execute(
146 "UPDATE tracks SET last_played = item_id WHERE item_id IN (2, 4, 6)"
147 )
148 await mass.music.database.commit()
149
150
151def _legacy_tracks_base_query() -> str:
152 """Return the legacy tracks base query (with the album_tracks JOIN)."""
153 return """
154 SELECT
155 tracks.*,
156 (SELECT JSON_GROUP_ARRAY(
157 json_object(
158 'item_id', track_pm.provider_item_id,
159 'provider_domain', track_pm.provider_domain,
160 'provider_instance', track_pm.provider_instance,
161 'available', track_pm.available,
162 'audio_format', json(track_pm.audio_format),
163 'url', track_pm.url,
164 'details', track_pm.details,
165 'in_library', track_pm.in_library,
166 'is_unique', track_pm.is_unique
167 )) FROM provider_mappings track_pm WHERE track_pm.item_id = tracks.item_id AND track_pm.media_type = 'track') AS provider_mappings,
168 (SELECT JSON_GROUP_ARRAY(
169 json_object(
170 'item_id', artists.item_id,
171 'provider', 'library',
172 'name', artists.name,
173 'sort_name', artists.sort_name,
174 'media_type', 'artist',
175 'external_ids', json((SELECT json_group_array(json_array(
176 external_id_lookup.external_id_type, external_id_lookup.external_id))
177 FROM external_id_lookup WHERE external_id_lookup.media_type = 'artist'
178 AND external_id_lookup.item_id = artists.item_id))
179 )) FROM artists JOIN track_artists on track_artists.track_id = tracks.item_id WHERE artists.item_id = track_artists.artist_id) AS artists,
180 (SELECT
181 json_object(
182 'item_id', albums.item_id,
183 'provider', 'library',
184 'name', albums.name,
185 'sort_name', albums.sort_name,
186 'media_type', 'album',
187 'year', albums.year,
188 'disc_number', album_tracks.disc_number,
189 'track_number', album_tracks.track_number,
190 'images', json_extract(albums.metadata, '$.images')
191 ) FROM albums WHERE albums.item_id = album_tracks.album_id) AS track_album
192 FROM tracks
193 LEFT JOIN album_tracks on album_tracks.track_id = tracks.item_id
194 """
195
196
197def _legacy_query( # noqa: PLR0913
198 controller: MediaControllerBase[Any],
199 *,
200 base_query: str | None = None,
201 favorite: bool | None = None,
202 search: str | None = None,
203 provider_filter: list[str] | None = None,
204 in_library_only: bool = False,
205 played_only: bool = False,
206 genre_ids: list[int] | None = None,
207 extra_query_parts: list[str] | None = None,
208 extra_join_parts: list[str] | None = None,
209 order_by: str | None = "sort_name",
210) -> tuple[str, dict[str, Any]]:
211 """Build a listing query the way the legacy (JOIN + GROUP BY) builder did."""
212 table = controller.db_table
213 query_parts: list[str] = list(extra_query_parts or [])
214 join_parts: list[str] = list(extra_join_parts or [])
215 params: dict[str, Any] = {}
216 if search:
217 query_parts.append(f"{table}.search_name LIKE :search")
218 params["search"] = f"%{create_safe_string(search, True, True)}%"
219 if favorite is not None:
220 query_parts.append(f"{table}.favorite = :favorite")
221 params["favorite"] = favorite
222 if played_only:
223 query_parts.append(f"{table}.last_played > 0")
224 if genre_ids:
225 params["genre_ids"] = genre_ids
226 params["genre_media_type"] = controller.media_type.value
227 query_parts.append(
228 f"EXISTS(SELECT 1 FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} gm "
229 f"WHERE gm.media_id = {table}.item_id "
230 "AND gm.media_type = :genre_media_type AND gm.genre_id IN :genre_ids)"
231 )
232 if provider_filter:
233 provider_conditions = []
234 for idx, prov in enumerate(provider_filter):
235 provider_conditions.append(f"provider_mappings.provider_instance = :prov_{idx}")
236 params[f"prov_{idx}"] = prov
237 params["provider_media_type"] = controller.media_type.value
238 in_library_clause = "AND provider_mappings.in_library = 1 " if in_library_only else ""
239 join_parts.append(
240 f"JOIN provider_mappings ON provider_mappings.item_id = {table}.item_id "
241 "AND provider_mappings.media_type = :provider_media_type "
242 f"{in_library_clause}"
243 f"AND ({' OR '.join(provider_conditions)})"
244 )
245 elif in_library_only:
246 params["provider_media_type"] = controller.media_type.value
247 join_parts.append(
248 f"JOIN provider_mappings ON provider_mappings.item_id = {table}.item_id "
249 "AND provider_mappings.media_type = :provider_media_type "
250 "AND provider_mappings.in_library = 1"
251 )
252 sql_query = base_query if base_query is not None else controller.base_query[0]
253 if join_parts:
254 sql_query += f" {' '.join(join_parts)} "
255 if query_parts:
256 sql_query += " WHERE " + " AND ".join(
257 x[5:] if x.lower().startswith("where ") else x for x in query_parts
258 )
259 sql_query += f" GROUP BY {table}.item_id"
260 if order_by:
261 sort_key = {
262 "sort_name": "search_sort_name ASC",
263 "sort_name_desc": "search_sort_name DESC",
264 "timestamp_added": "timestamp_added ASC",
265 "timestamp_added_desc": "timestamp_added DESC",
266 "last_played": "last_played ASC",
267 "last_played_desc": "last_played DESC",
268 }[order_by]
269 sql_query += f" ORDER BY {sort_key}"
270 return sql_query, params
271
272
273async def _compare(
274 mass: MusicAssistant,
275 controller: MediaControllerBase[Any],
276 *,
277 legacy_base_query: str | None = None,
278 limit: int = 500,
279 offset: int = 0,
280 **filter_kwargs: Any,
281) -> list[Any]:
282 """Run the new and legacy listing queries and assert identical items/order."""
283 legacy_sql, legacy_params = _legacy_query(
284 controller, base_query=legacy_base_query, **filter_kwargs
285 )
286 legacy_rows = await mass.music.database.get_rows_from_query(
287 legacy_sql, legacy_params, limit=limit, offset=offset
288 )
289 order_by = filter_kwargs.pop("order_by", "sort_name")
290 new_items = await controller.get_library_items_by_query(
291 order_by=order_by, limit=limit, offset=offset, **filter_kwargs
292 )
293 assert [str(row["item_id"]) for row in legacy_rows] == [x.item_id for x in new_items]
294 # provider mappings must be hydrated identically
295 for row, item in zip(legacy_rows, new_items, strict=True):
296 legacy_mappings = {
297 (m["provider_instance"], m["item_id"]) for m in json.loads(row["provider_mappings"])
298 }
299 new_mappings = {(m.provider_instance, m.item_id) for m in item.provider_mappings}
300 assert legacy_mappings == new_mappings
301 assert isinstance(new_items, list)
302 return new_items
303
304
305FILTER_MATRIX: list[dict[str, Any]] = [
306 {"in_library_only": True},
307 {"in_library_only": True, "favorite": True},
308 {"in_library_only": True, "search": "02"},
309 {"in_library_only": True, "search": "rack"},
310 {"in_library_only": True, "search": "Track 02"},
311 {"in_library_only": True, "provider_filter": ["prov_b_inst"]},
312 {"in_library_only": True, "provider_filter": ["prov_a_inst", "prov_b_inst"]},
313 {"provider_filter": ["prov_b_inst"]},
314 {"in_library_only": True, "favorite": False, "search": "0", "order_by": "timestamp_added"},
315 {
316 "in_library_only": True,
317 "favorite": True,
318 "provider_filter": ["prov_a_inst"],
319 "search": "1",
320 },
321 {"in_library_only": True, "order_by": "sort_name_desc"},
322 {"in_library_only": True, "order_by": "timestamp_added_desc"},
323]
324
325
326@pytest.mark.parametrize("filter_kwargs", FILTER_MATRIX)
327async def test_artist_listing_matches_legacy(
328 seeded_mass: MusicAssistant, filter_kwargs: dict[str, Any]
329) -> None:
330 """Artist listings return the same items/order as the legacy JOIN queries."""
331 await _compare(seeded_mass, seeded_mass.music.artists, **dict(filter_kwargs))
332
333
334@pytest.mark.parametrize("filter_kwargs", FILTER_MATRIX)
335async def test_album_listing_matches_legacy(
336 seeded_mass: MusicAssistant, filter_kwargs: dict[str, Any]
337) -> None:
338 """Album listings return the same items/order as the legacy JOIN queries."""
339 await _compare(seeded_mass, seeded_mass.music.albums, **dict(filter_kwargs))
340
341
342@pytest.mark.parametrize("filter_kwargs", FILTER_MATRIX)
343async def test_track_listing_matches_legacy(
344 seeded_mass: MusicAssistant, filter_kwargs: dict[str, Any]
345) -> None:
346 """Track listings return the same items/order as the legacy JOIN queries."""
347 await _compare(
348 seeded_mass,
349 seeded_mass.music.tracks,
350 legacy_base_query=_legacy_tracks_base_query(),
351 **dict(filter_kwargs),
352 )
353
354
355async def test_track_listing_with_pagination_matches_legacy(seeded_mass: MusicAssistant) -> None:
356 """Paged track listings return the same items/order as the legacy queries."""
357 for offset in (0, 5, 15):
358 await _compare(
359 seeded_mass,
360 seeded_mass.music.tracks,
361 legacy_base_query=_legacy_tracks_base_query(),
362 limit=5,
363 offset=offset,
364 in_library_only=True,
365 )
366
367
368async def test_track_listing_played_only_matches_legacy(seeded_mass: MusicAssistant) -> None:
369 """played_only track listings return the same items as the legacy queries."""
370 items = await _compare(
371 seeded_mass,
372 seeded_mass.music.tracks,
373 legacy_base_query=_legacy_tracks_base_query(),
374 in_library_only=True,
375 played_only=True,
376 )
377 assert len(items) == 3
378
379
380async def test_track_listing_genre_filter_matches_legacy(seeded_mass: MusicAssistant) -> None:
381 """Genre-filtered track listings return the same items as the legacy queries."""
382 items = await _compare(
383 seeded_mass,
384 seeded_mass.music.tracks,
385 legacy_base_query=_legacy_tracks_base_query(),
386 in_library_only=True,
387 genre_ids=[1],
388 )
389 assert len(items) == 3
390
391
392async def test_track_listing_artist_join_matches_legacy(seeded_mass: MusicAssistant) -> None:
393 """Artist-name search (with join fanout) still dedupes via GROUP BY."""
394 join = (
395 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
396 "JOIN artists ON artists.item_id = track_artists.artist_id "
397 "AND artists.search_name LIKE :search_artist"
398 )
399 legacy_sql, legacy_params = _legacy_query(
400 seeded_mass.music.tracks,
401 base_query=_legacy_tracks_base_query(),
402 in_library_only=True,
403 extra_join_parts=[join],
404 )
405 legacy_params["search_artist"] = "%artist02%"
406 legacy_rows = await seeded_mass.music.database.get_rows_from_query(legacy_sql, legacy_params)
407 new_items = await seeded_mass.music.tracks.get_library_items_by_query(
408 order_by="sort_name",
409 in_library_only=True,
410 extra_join_parts=[join],
411 extra_query_params={"search_artist": "%artist02%"},
412 )
413 assert [str(row["item_id"]) for row in legacy_rows] == [x.item_id for x in new_items]
414 assert len(new_items) > 0
415
416
417async def test_track_artist_name_sorting(seeded_mass: MusicAssistant) -> None:
418 """Tracks can be sorted by artist name, with secondary sorting by track name."""
419 # Test ascending artist name sort
420 tracks_asc = await seeded_mass.music.tracks.library_items(order_by="track_artist_name")
421 # Build list of (artist_name, track_name) tuples for validation
422 artist_track_pairs_asc = [(t.artists[0].name, t.name) for t in tracks_asc if t.artists]
423 # Sort should be by artist name first, then track name
424 expected_asc = sorted(artist_track_pairs_asc, key=lambda x: (x[0], x[1]))
425 assert artist_track_pairs_asc == expected_asc, "Should sort by artist name, then track name"
426 assert len(artist_track_pairs_asc) > 0, "Should return tracks with artists"
427
428 # Test descending artist name sort
429 tracks_desc = await seeded_mass.music.tracks.library_items(order_by="track_artist_name_desc")
430 artist_track_pairs_desc = [(t.artists[0].name, t.name) for t in tracks_desc if t.artists]
431 # DESC sort: artist name descending, but track name still ascending within each artist
432 grouped: dict[str, list[str]] = {}
433 for pair in artist_track_pairs_asc:
434 artist, track = pair
435 if artist not in grouped:
436 grouped[artist] = []
437 grouped[artist].append(track)
438 # Sort artists descending, tracks within artist ascending
439 expected_desc = []
440 for artist in sorted(grouped.keys(), reverse=True):
441 for track in sorted(grouped[artist]):
442 expected_desc.append((artist, track))
443 assert artist_track_pairs_desc == expected_desc, "Should sort by artist name descending"
444
445
446async def test_album_artist_name_sorting(seeded_mass: MusicAssistant) -> None:
447 """Albums can be sorted by artist name, with secondary sorting by year."""
448 # Test ascending artist name sort
449 albums_asc = await seeded_mass.music.albums.library_items(order_by="album_artist_name")
450 # Build list of (artist_name, album_name, year) tuples for validation
451 artist_album_pairs_asc = [(a.artists[0].name, a.name, a.year) for a in albums_asc if a.artists]
452 # Sort should be by artist name first, then year descending
453 expected_asc = sorted(artist_album_pairs_asc, key=lambda x: (x[0], -x[2] if x[2] else 0))
454 assert artist_album_pairs_asc == expected_asc, "Should sort by artist name, then year desc"
455 assert len(artist_album_pairs_asc) > 0, "Should return albums with artists"
456
457 # Test descending artist name sort
458 albums_desc = await seeded_mass.music.albums.library_items(order_by="album_artist_name_desc")
459 artist_album_pairs_desc = [
460 (a.artists[0].name, a.name, a.year) for a in albums_desc if a.artists
461 ]
462 # SQL: artists.search_name DESC, year DESC - both descending
463 expected_desc = sorted(artist_album_pairs_asc, key=lambda x: x[2] or 0, reverse=True)
464 expected_desc = sorted(expected_desc, key=lambda x: x[0], reverse=True)
465 assert artist_album_pairs_desc == expected_desc, "Should sort by artist name descending"
466
467
468async def test_hidden_track_excluded_from_library_listing(seeded_mass: MusicAssistant) -> None:
469 """A track whose only mapping has in_library=0 is hidden from library listings."""
470 items = await seeded_mass.music.tracks.get_library_items_by_query(in_library_only=True)
471 assert "Track hidden" not in {x.name for x in items}
472 # ...but still returned when not filtering on library membership
473 items = await seeded_mass.music.tracks.get_library_items_by_query()
474 assert "Track hidden" in {x.name for x in items}
475
476
477async def test_random_order_applies_filters(seeded_mass: MusicAssistant) -> None:
478 """The random-order subquery applies the provider/in-library filters."""
479 items = await seeded_mass.music.tracks.get_library_items_by_query(
480 order_by="random", in_library_only=True, provider_filter=["prov_b_inst"]
481 )
482 expected = await seeded_mass.music.tracks.get_library_items_by_query(
483 in_library_only=True, provider_filter=["prov_b_inst"]
484 )
485 assert {x.item_id for x in items} == {x.item_id for x in expected}
486
487
488async def test_album_tracks_returns_album_scoped_disc_and_track_numbers(
489 seeded_mass: MusicAssistant,
490) -> None:
491 """get_library_album_tracks returns the requested album's disc/track numbers."""
492 # track 3 is on two albums with different disc/track numbers
493 row = await seeded_mass.music.database.get_row("tracks", {"name": "Track 03"})
494 assert row is not None
495 track_db_id = int(row["item_id"])
496 album_track_rows = await seeded_mass.music.database.get_rows(
497 DB_TABLE_ALBUM_TRACKS, {"track_id": track_db_id}
498 )
499 assert len(album_track_rows) == 2
500 for album_track_row in album_track_rows:
501 album_tracks = await seeded_mass.music.albums.get_library_album_tracks(
502 album_track_row["album_id"]
503 )
504 track = next(x for x in album_tracks if int(x.item_id) == track_db_id)
505 assert track.album is not None
506 assert int(track.album.item_id) == album_track_row["album_id"]
507 assert track.disc_number == album_track_row["disc_number"]
508 assert track.track_number == album_track_row["track_number"]
509
510
511async def test_album_tracks_matches_legacy(seeded_mass: MusicAssistant) -> None:
512 """get_library_album_tracks returns the same tracks as the legacy album JOIN query."""
513 albums = await seeded_mass.music.albums.get_library_items_by_query(in_library_only=True)
514 for album in albums:
515 legacy_sql, legacy_params = _legacy_query(
516 seeded_mass.music.tracks,
517 base_query=_legacy_tracks_base_query(),
518 extra_query_parts=["WHERE album_tracks.album_id = :album_id"],
519 order_by=None,
520 )
521 legacy_params["album_id"] = int(album.item_id)
522 legacy_rows = await seeded_mass.music.database.get_rows_from_query(
523 legacy_sql, legacy_params
524 )
525 new_items = await seeded_mass.music.albums.get_library_album_tracks(album.item_id)
526 assert sorted(str(row["item_id"]) for row in legacy_rows) == sorted(
527 x.item_id for x in new_items
528 )
529
530
531async def test_audiobook_listing_resume_info(seeded_mass: MusicAssistant) -> None:
532 """Audiobook listings dedupe playlog rows and surface the most recent resume info."""
533 audiobook = Audiobook(
534 item_id="0",
535 provider="library",
536 name="Book 01",
537 provider_mappings={_mapping()},
538 )
539 db_book = await seeded_mass.music.audiobooks.add_item_to_library(audiobook)
540 # two playlog rows for the same book (e.g. two users/providers)
541 for userid, timestamp, seconds_played in (("user-a", 100, 60), ("user-b", 200, 120)):
542 await seeded_mass.music.database.insert(
543 DB_TABLE_PLAYLOG,
544 {
545 "item_id": db_book.item_id,
546 "provider": "library",
547 "media_type": MediaType.AUDIOBOOK.value,
548 "name": db_book.name,
549 "timestamp": timestamp,
550 "fully_played": False,
551 "seconds_played": seconds_played,
552 "userid": userid,
553 "user_initiated": True,
554 },
555 )
556 await seeded_mass.music.database.commit()
557 items = await seeded_mass.music.audiobooks.get_library_items_by_query(in_library_only=True)
558 books = [x for x in items if x.item_id == db_book.item_id]
559 # the playlog join may not fan out the result set
560 assert len(books) == 1
561 # the most recent playlog entry wins
562 assert books[0].resume_position_ms == 120 * 1000
563
564
565async def test_audiobook_collections_collapse_and_preserve_order(
566 mass: MusicAssistant,
567) -> None:
568 """Collapsed audiobook collections remain ordered and reuse the cached row shape."""
569 controller = mass.music.audiobooks
570 assert (
571 await controller.get_library_items_by_query(
572 in_library_only=True,
573 collapse_collections=True,
574 )
575 == []
576 )
577
578 collection_books = (
579 ("Alpha 2", "Alpha Series", 2.0),
580 ("Alpha 1.5", "Alpha Series", "1.5"),
581 ("Alpha 1", "Alpha Series", 1.0),
582 ("Beta 1", "Beta Series", 1.0),
583 )
584 for name, collection_name, sequence in collection_books:
585 await controller.add_item_to_library(
586 Audiobook(
587 item_id="0",
588 provider="library",
589 name=name,
590 provider_mappings={_mapping()},
591 metadata=MediaItemMetadata(
592 collections=UniqueList(
593 [MediaItemCollection(title=collection_name, sequence=sequence)]
594 )
595 ),
596 )
597 )
598 await controller.add_item_to_library(
599 Audiobook(
600 item_id="0",
601 provider="library",
602 name="Standalone",
603 provider_mappings={_mapping()},
604 )
605 )
606
607 items = await controller.get_library_items_by_query(
608 in_library_only=True,
609 order_by="name_desc",
610 collapse_collections=True,
611 )
612 collections = {item.name: item for item in items if isinstance(item, MediaCollection)}
613 assert list(collections) == ["Beta Series", "Alpha Series"]
614 assert [item.name for item in collections["Alpha Series"].items] == [
615 "Alpha 1",
616 "Alpha 1.5",
617 "Alpha 2",
618 ]
619
620 with patch.object(
621 mass.music.database,
622 "get_rows_from_query",
623 wraps=mass.music.database.get_rows_from_query,
624 ) as get_rows:
625 search_results = await controller.get_library_items_by_query(
626 search="Series",
627 in_library_only=True,
628 order_by="name_desc",
629 collapse_collections=True,
630 )
631 assert [item.name for item in search_results] == ["Beta Series", "Alpha Series"]
632 assert get_rows.await_count == 1
633
634 collection = await controller.get_collection(collections["Alpha Series"].item_id)
635 assert [item.name for item in collection.items] == ["Alpha 1", "Alpha 1.5", "Alpha 2"]
636 assert all(isinstance(item, Audiobook) for item in collection.items)
637
638
639async def test_artist_audiobooks_collapse_collections(
640 mass: MusicAssistant,
641) -> None:
642 """Artist audiobook listings collapse collections while preserving the artist scope."""
643 author = Artist(
644 item_id="0",
645 provider="library",
646 name="Test Author",
647 provider_mappings={_mapping()},
648 artist_type=ArtistType.AUTHOR,
649 )
650 author = await mass.music.artists.add_item_to_library(author)
651
652 for name, sequence in (("Book 2", 2), ("Book 1", 1)):
653 await mass.music.audiobooks.add_item_to_library(
654 Audiobook(
655 item_id="0",
656 provider="library",
657 name=name,
658 provider_mappings={_mapping()},
659 authors=UniqueList([author]),
660 metadata=MediaItemMetadata(
661 collections=UniqueList(
662 [MediaItemCollection(title="Test Collection", sequence=sequence)]
663 )
664 ),
665 )
666 )
667
668 await mass.music.audiobooks.add_item_to_library(
669 Audiobook(
670 item_id="0",
671 provider="library",
672 name="Standalone",
673 provider_mappings={_mapping()},
674 authors=UniqueList([author]),
675 )
676 )
677
678 result = await mass.music.artists.audiobooks(
679 author.item_id,
680 author.provider,
681 author.artist_type,
682 in_library_only=True,
683 collapse_collections=True,
684 )
685
686 assert len(result) == 2
687
688 collection = next(item for item in result if isinstance(item, MediaCollection))
689 standalone = next(item for item in result if isinstance(item, Audiobook))
690
691 assert collection.name == "Test Collection"
692 assert [item.name for item in collection.items] == ["Book 1", "Book 2"]
693 assert standalone.name == "Standalone"
694
695
696async def test_listing_queries_stream_from_sort_index(seeded_mass: MusicAssistant) -> None:
697 """Default library listings stream from the sort index without a temp b-tree."""
698 database = seeded_mass.music.database
699 # drop the planner statistics so SQLite falls back to its default (large) table
700 # size estimates: with only a handful of seeded rows it would (rightfully) just
701 # scan the table, while the streaming-from-sort-index behaviour is what matters
702 # for large libraries
703 await database.execute("ANALYZE")
704 await database.execute("DELETE FROM sqlite_stat1")
705 await database.execute("ANALYZE sqlite_master")
706 try:
707 for controller in (
708 seeded_mass.music.artists,
709 seeded_mass.music.albums,
710 seeded_mass.music.tracks,
711 ):
712 captured: dict[str, Any] = {}
713 orig = database.get_rows_from_query
714
715 async def spy(
716 query: str,
717 params: dict[str, Any] | None = None,
718 limit: int = 500,
719 offset: int = 0,
720 _captured: dict[str, Any] = captured,
721 _orig: Any = orig,
722 ) -> list[Any]:
723 _captured["query"], _captured["params"] = query, params
724 result: list[Any] = await _orig(query, params, limit=limit, offset=offset)
725 return result
726
727 with patch.object(database, "get_rows_from_query", spy):
728 await controller.get_library_items_by_query(
729 order_by="sort_name", in_library_only=True
730 )
731 plan_rows = await database.get_rows_from_query(
732 f"EXPLAIN QUERY PLAN {captured['query']} LIMIT 500 OFFSET 0",
733 captured["params"],
734 limit=0,
735 )
736 details = [row["detail"] for row in plan_rows]
737 table = controller.db_table
738 assert any(
739 f"SCAN {table} USING INDEX {table}_search_sort_name_idx" in detail
740 for detail in details
741 ), details
742 # no temp b-tree may be used to sort the (potentially huge) main result;
743 # correlated subqueries only sort the few rows of a single item
744 top_level = [row["detail"] for row in plan_rows if row["parent"] == 0]
745 assert not any("TEMP B-TREE" in detail.upper() for detail in top_level), details
746 finally:
747 # restore the real statistics for any tests running after this one
748 await database.execute("ANALYZE")
749
750
751async def test_search_uses_fts_index(seeded_mass: MusicAssistant) -> None:
752 """Search terms of >= 3 chars are matched through the FTS index."""
753 database = seeded_mass.music.database
754 captured: dict[str, Any] = {}
755 orig = database.get_rows_from_query
756
757 async def spy(
758 query: str,
759 params: dict[str, Any] | None = None,
760 limit: int = 500,
761 offset: int = 0,
762 ) -> list[Any]:
763 captured["query"], captured["params"] = query, params
764 return await orig(query, params, limit=limit, offset=offset)
765
766 with patch.object(database, "get_rows_from_query", spy):
767 items = await seeded_mass.music.tracks.get_library_items_by_query(
768 search="rack", in_library_only=True
769 )
770 assert "tracks_fts MATCH" in captured["query"]
771 # midword substring matching works the same as the LIKE scan did
772 assert len(items) == 20
773
774 # terms shorter than a trigram fall back to a LIKE scan
775 with patch.object(database, "get_rows_from_query", spy):
776 items = await seeded_mass.music.tracks.get_library_items_by_query(
777 search="02", in_library_only=True
778 )
779 assert "tracks_fts" not in captured["query"]
780 assert "search_name LIKE" in captured["query"]
781 assert {x.name for x in items} == {"Track 02"}
782
783
784async def test_fts_index_follows_item_changes(seeded_mass: MusicAssistant) -> None:
785 """The FTS index reflects added, renamed and deleted library items."""
786 database = seeded_mass.music.database
787 artists = await seeded_mass.music.artists.get_library_items_by_query(limit=1)
788 track = Track(
789 item_id="0",
790 provider="library",
791 name="Bohemian Rhapsody",
792 provider_mappings={_mapping()},
793 artists=UniqueList([artists[0]]),
794 )
795 db_track = await seeded_mass.music.tracks.add_item_to_library(track)
796
797 async def _search(term: str) -> set[str]:
798 items = await seeded_mass.music.tracks.get_library_items_by_query(search=term)
799 return {x.item_id for x in items}
800
801 assert db_track.item_id in await _search("rhapsody")
802
803 # a rename must be picked up by the index
804 await database.execute(
805 "UPDATE tracks SET name = 'Radio Ga Ga', search_name = 'radiogaga' "
806 "WHERE item_id = :item_id",
807 {"item_id": int(db_track.item_id)},
808 )
809 await database.commit()
810 assert db_track.item_id not in await _search("rhapsody")
811 assert db_track.item_id in await _search("gaga")
812
813 # a delete must remove the item from the index
814 await database.execute(
815 "DELETE FROM tracks WHERE item_id = :item_id", {"item_id": int(db_track.item_id)}
816 )
817 await database.commit()
818 assert db_track.item_id not in await _search("gaga")
819