music-assistant-server

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_hidden_track_excluded_from_library_listing(seeded_mass: MusicAssistant) -> None:
418    """A track whose only mapping has in_library=0 is hidden from library listings."""
419    items = await seeded_mass.music.tracks.get_library_items_by_query(in_library_only=True)
420    assert "Track hidden" not in {x.name for x in items}
421    # ...but still returned when not filtering on library membership
422    items = await seeded_mass.music.tracks.get_library_items_by_query()
423    assert "Track hidden" in {x.name for x in items}
424
425
426async def test_random_order_applies_filters(seeded_mass: MusicAssistant) -> None:
427    """The random-order subquery applies the provider/in-library filters."""
428    items = await seeded_mass.music.tracks.get_library_items_by_query(
429        order_by="random", in_library_only=True, provider_filter=["prov_b_inst"]
430    )
431    expected = await seeded_mass.music.tracks.get_library_items_by_query(
432        in_library_only=True, provider_filter=["prov_b_inst"]
433    )
434    assert {x.item_id for x in items} == {x.item_id for x in expected}
435
436
437async def test_album_tracks_returns_album_scoped_disc_and_track_numbers(
438    seeded_mass: MusicAssistant,
439) -> None:
440    """get_library_album_tracks returns the requested album's disc/track numbers."""
441    # track 3 is on two albums with different disc/track numbers
442    row = await seeded_mass.music.database.get_row("tracks", {"name": "Track 03"})
443    assert row is not None
444    track_db_id = int(row["item_id"])
445    album_track_rows = await seeded_mass.music.database.get_rows(
446        DB_TABLE_ALBUM_TRACKS, {"track_id": track_db_id}
447    )
448    assert len(album_track_rows) == 2
449    for album_track_row in album_track_rows:
450        album_tracks = await seeded_mass.music.albums.get_library_album_tracks(
451            album_track_row["album_id"]
452        )
453        track = next(x for x in album_tracks if int(x.item_id) == track_db_id)
454        assert track.album is not None
455        assert int(track.album.item_id) == album_track_row["album_id"]
456        assert track.disc_number == album_track_row["disc_number"]
457        assert track.track_number == album_track_row["track_number"]
458
459
460async def test_album_tracks_matches_legacy(seeded_mass: MusicAssistant) -> None:
461    """get_library_album_tracks returns the same tracks as the legacy album JOIN query."""
462    albums = await seeded_mass.music.albums.get_library_items_by_query(in_library_only=True)
463    for album in albums:
464        legacy_sql, legacy_params = _legacy_query(
465            seeded_mass.music.tracks,
466            base_query=_legacy_tracks_base_query(),
467            extra_query_parts=["WHERE album_tracks.album_id = :album_id"],
468            order_by=None,
469        )
470        legacy_params["album_id"] = int(album.item_id)
471        legacy_rows = await seeded_mass.music.database.get_rows_from_query(
472            legacy_sql, legacy_params
473        )
474        new_items = await seeded_mass.music.albums.get_library_album_tracks(album.item_id)
475        assert sorted(str(row["item_id"]) for row in legacy_rows) == sorted(
476            x.item_id for x in new_items
477        )
478
479
480async def test_audiobook_listing_resume_info(seeded_mass: MusicAssistant) -> None:
481    """Audiobook listings dedupe playlog rows and surface the most recent resume info."""
482    audiobook = Audiobook(
483        item_id="0",
484        provider="library",
485        name="Book 01",
486        provider_mappings={_mapping()},
487    )
488    db_book = await seeded_mass.music.audiobooks.add_item_to_library(audiobook)
489    # two playlog rows for the same book (e.g. two users/providers)
490    for userid, timestamp, seconds_played in (("user-a", 100, 60), ("user-b", 200, 120)):
491        await seeded_mass.music.database.insert(
492            DB_TABLE_PLAYLOG,
493            {
494                "item_id": db_book.item_id,
495                "provider": "library",
496                "media_type": MediaType.AUDIOBOOK.value,
497                "name": db_book.name,
498                "timestamp": timestamp,
499                "fully_played": False,
500                "seconds_played": seconds_played,
501                "userid": userid,
502                "user_initiated": True,
503            },
504        )
505    await seeded_mass.music.database.commit()
506    items = await seeded_mass.music.audiobooks.get_library_items_by_query(in_library_only=True)
507    books = [x for x in items if x.item_id == db_book.item_id]
508    # the playlog join may not fan out the result set
509    assert len(books) == 1
510    # the most recent playlog entry wins
511    assert books[0].resume_position_ms == 120 * 1000
512
513
514async def test_audiobook_collections_collapse_and_preserve_order(
515    mass: MusicAssistant,
516) -> None:
517    """Collapsed audiobook collections remain ordered and reuse the cached row shape."""
518    controller = mass.music.audiobooks
519    assert (
520        await controller.get_library_items_by_query(
521            in_library_only=True,
522            collapse_collections=True,
523        )
524        == []
525    )
526
527    collection_books = (
528        ("Alpha 2", "Alpha Series", 2.0),
529        ("Alpha 1.5", "Alpha Series", "1.5"),
530        ("Alpha 1", "Alpha Series", 1.0),
531        ("Beta 1", "Beta Series", 1.0),
532    )
533    for name, collection_name, sequence in collection_books:
534        await controller.add_item_to_library(
535            Audiobook(
536                item_id="0",
537                provider="library",
538                name=name,
539                provider_mappings={_mapping()},
540                metadata=MediaItemMetadata(
541                    collections=UniqueList(
542                        [MediaItemCollection(title=collection_name, sequence=sequence)]
543                    )
544                ),
545            )
546        )
547    await controller.add_item_to_library(
548        Audiobook(
549            item_id="0",
550            provider="library",
551            name="Standalone",
552            provider_mappings={_mapping()},
553        )
554    )
555
556    items = await controller.get_library_items_by_query(
557        in_library_only=True,
558        order_by="name_desc",
559        collapse_collections=True,
560    )
561    collections = {item.name: item for item in items if isinstance(item, MediaCollection)}
562    assert list(collections) == ["Beta Series", "Alpha Series"]
563    assert [item.name for item in collections["Alpha Series"].items] == [
564        "Alpha 1",
565        "Alpha 1.5",
566        "Alpha 2",
567    ]
568
569    with patch.object(
570        mass.music.database,
571        "get_rows_from_query",
572        wraps=mass.music.database.get_rows_from_query,
573    ) as get_rows:
574        search_results = await controller.get_library_items_by_query(
575            search="Series",
576            in_library_only=True,
577            order_by="name_desc",
578            collapse_collections=True,
579        )
580    assert [item.name for item in search_results] == ["Beta Series", "Alpha Series"]
581    assert get_rows.await_count == 1
582
583    collection = await controller.get_collection(collections["Alpha Series"].item_id)
584    assert [item.name for item in collection.items] == ["Alpha 1", "Alpha 1.5", "Alpha 2"]
585    assert all(isinstance(item, Audiobook) for item in collection.items)
586
587
588async def test_artist_audiobooks_collapse_collections(
589    mass: MusicAssistant,
590) -> None:
591    """Artist audiobook listings collapse collections while preserving the artist scope."""
592    author = Artist(
593        item_id="0",
594        provider="library",
595        name="Test Author",
596        provider_mappings={_mapping()},
597        artist_type=ArtistType.AUTHOR,
598    )
599    author = await mass.music.artists.add_item_to_library(author)
600
601    for name, sequence in (("Book 2", 2), ("Book 1", 1)):
602        await mass.music.audiobooks.add_item_to_library(
603            Audiobook(
604                item_id="0",
605                provider="library",
606                name=name,
607                provider_mappings={_mapping()},
608                authors=UniqueList([author]),
609                metadata=MediaItemMetadata(
610                    collections=UniqueList(
611                        [MediaItemCollection(title="Test Collection", sequence=sequence)]
612                    )
613                ),
614            )
615        )
616
617    await mass.music.audiobooks.add_item_to_library(
618        Audiobook(
619            item_id="0",
620            provider="library",
621            name="Standalone",
622            provider_mappings={_mapping()},
623            authors=UniqueList([author]),
624        )
625    )
626
627    result = await mass.music.artists.audiobooks(
628        author.item_id,
629        author.provider,
630        author.artist_type,
631        in_library_only=True,
632        collapse_collections=True,
633    )
634
635    assert len(result) == 2
636
637    collection = next(item for item in result if isinstance(item, MediaCollection))
638    standalone = next(item for item in result if isinstance(item, Audiobook))
639
640    assert collection.name == "Test Collection"
641    assert [item.name for item in collection.items] == ["Book 1", "Book 2"]
642    assert standalone.name == "Standalone"
643
644
645async def test_listing_queries_stream_from_sort_index(seeded_mass: MusicAssistant) -> None:
646    """Default library listings stream from the sort index without a temp b-tree."""
647    database = seeded_mass.music.database
648    # drop the planner statistics so SQLite falls back to its default (large) table
649    # size estimates: with only a handful of seeded rows it would (rightfully) just
650    # scan the table, while the streaming-from-sort-index behaviour is what matters
651    # for large libraries
652    await database.execute("ANALYZE")
653    await database.execute("DELETE FROM sqlite_stat1")
654    await database.execute("ANALYZE sqlite_master")
655    try:
656        for controller in (
657            seeded_mass.music.artists,
658            seeded_mass.music.albums,
659            seeded_mass.music.tracks,
660        ):
661            captured: dict[str, Any] = {}
662            orig = database.get_rows_from_query
663
664            async def spy(
665                query: str,
666                params: dict[str, Any] | None = None,
667                limit: int = 500,
668                offset: int = 0,
669                _captured: dict[str, Any] = captured,
670                _orig: Any = orig,
671            ) -> list[Any]:
672                _captured["query"], _captured["params"] = query, params
673                result: list[Any] = await _orig(query, params, limit=limit, offset=offset)
674                return result
675
676            with patch.object(database, "get_rows_from_query", spy):
677                await controller.get_library_items_by_query(
678                    order_by="sort_name", in_library_only=True
679                )
680            plan_rows = await database.get_rows_from_query(
681                f"EXPLAIN QUERY PLAN {captured['query']} LIMIT 500 OFFSET 0",
682                captured["params"],
683                limit=0,
684            )
685            details = [row["detail"] for row in plan_rows]
686            table = controller.db_table
687            assert any(
688                f"SCAN {table} USING INDEX {table}_search_sort_name_idx" in detail
689                for detail in details
690            ), details
691            # no temp b-tree may be used to sort the (potentially huge) main result;
692            # correlated subqueries only sort the few rows of a single item
693            top_level = [row["detail"] for row in plan_rows if row["parent"] == 0]
694            assert not any("TEMP B-TREE" in detail.upper() for detail in top_level), details
695    finally:
696        # restore the real statistics for any tests running after this one
697        await database.execute("ANALYZE")
698
699
700async def test_search_uses_fts_index(seeded_mass: MusicAssistant) -> None:
701    """Search terms of >= 3 chars are matched through the FTS index."""
702    database = seeded_mass.music.database
703    captured: dict[str, Any] = {}
704    orig = database.get_rows_from_query
705
706    async def spy(
707        query: str,
708        params: dict[str, Any] | None = None,
709        limit: int = 500,
710        offset: int = 0,
711    ) -> list[Any]:
712        captured["query"], captured["params"] = query, params
713        return await orig(query, params, limit=limit, offset=offset)
714
715    with patch.object(database, "get_rows_from_query", spy):
716        items = await seeded_mass.music.tracks.get_library_items_by_query(
717            search="rack", in_library_only=True
718        )
719    assert "tracks_fts MATCH" in captured["query"]
720    # midword substring matching works the same as the LIKE scan did
721    assert len(items) == 20
722
723    # terms shorter than a trigram fall back to a LIKE scan
724    with patch.object(database, "get_rows_from_query", spy):
725        items = await seeded_mass.music.tracks.get_library_items_by_query(
726            search="02", in_library_only=True
727        )
728    assert "tracks_fts" not in captured["query"]
729    assert "search_name LIKE" in captured["query"]
730    assert {x.name for x in items} == {"Track 02"}
731
732
733async def test_fts_index_follows_item_changes(seeded_mass: MusicAssistant) -> None:
734    """The FTS index reflects added, renamed and deleted library items."""
735    database = seeded_mass.music.database
736    artists = await seeded_mass.music.artists.get_library_items_by_query(limit=1)
737    track = Track(
738        item_id="0",
739        provider="library",
740        name="Bohemian Rhapsody",
741        provider_mappings={_mapping()},
742        artists=UniqueList([artists[0]]),
743    )
744    db_track = await seeded_mass.music.tracks.add_item_to_library(track)
745
746    async def _search(term: str) -> set[str]:
747        items = await seeded_mass.music.tracks.get_library_items_by_query(search=term)
748        return {x.item_id for x in items}
749
750    assert db_track.item_id in await _search("rhapsody")
751
752    # a rename must be picked up by the index
753    await database.execute(
754        "UPDATE tracks SET name = 'Radio Ga Ga', search_name = 'radiogaga' "
755        "WHERE item_id = :item_id",
756        {"item_id": int(db_track.item_id)},
757    )
758    await database.commit()
759    assert db_track.item_id not in await _search("rhapsody")
760    assert db_track.item_id in await _search("gaga")
761
762    # a delete must remove the item from the index
763    await database.execute(
764        "DELETE FROM tracks WHERE item_id = :item_id", {"item_id": int(db_track.item_id)}
765    )
766    await database.commit()
767    assert db_track.item_id not in await _search("gaga")
768