import sqlite3 import tempfile import unittest from pathlib import Path from music_server.services import catalog_reader as catalog_reader_module from music_server.services.catalog_reader import CatalogReader class CatalogReaderTests(unittest.TestCase): def setUp(self) -> None: self._tmpdir = tempfile.TemporaryDirectory() self._db_path = Path(self._tmpdir.name) / "catalog_read.db" conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_playlists ( playlist_id integer primary key, platform text not null, remote_playlist_id text not null, name text not null, description text, cover_url text, play_count integer not null, song_count integer not null, playable_song_count integer not null ); create table catalog_tracks ( song_id integer primary key, platform text not null, remote_song_id text not null, name text not null, singers text, album text, cover_url text, duration_ms integer, metadata_json text ); create table catalog_track_files ( song_id integer not null, quality_label text not null, ext text not null, file_size_bytes integer, backend_type text not null, backend_name text not null, locator text not null, public_url text, status text not null, is_primary integer not null ); """ ) conn.commit() conn.close() def tearDown(self) -> None: self._tmpdir.cleanup() def _insert_playlists(self, rows: list[tuple]) -> None: conn = sqlite3.connect(self._db_path) conn.executemany( """ insert into catalog_playlists ( playlist_id, platform, remote_playlist_id, name, description, cover_url, play_count, song_count, playable_song_count ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, rows, ) conn.commit() conn.close() def test_list_playlists_filters_song_count_and_orders_by_play_count_then_id(self): self._insert_playlists( [ (1, "netease", "18165", "A", "desc", "https://img/a.jpg", 50, 3, 2), (2, "qq", "75", "B", "desc", "https://img/b.jpg", 120, 0, 0), (3, "qq", "99", "C", "desc", "https://img/c.jpg", 50, 2, 1), (4, "netease", "20", "D", "desc", "https://img/d.jpg", 200, 1, 1), ] ) reader = CatalogReader(db_path=str(self._db_path)) rows = reader.list_playlists(page=1, page_size=10) self.assertEqual([4, 1, 3], [row["playlist_id"] for row in rows]) self.assertEqual([1, 2, 1], [row["playable_song_count"] for row in rows]) def test_list_playlists_normalizes_page_and_page_size(self): self._insert_playlists( [ (1, "netease", "1", "A", "desc", "https://img/a.jpg", 10, 1, 1), (2, "qq", "2", "B", "desc", "https://img/b.jpg", 20, 1, 1), (3, "qq", "3", "C", "desc", "https://img/c.jpg", 30, 1, 1), ] ) reader = CatalogReader(db_path=str(self._db_path)) first_page = reader.list_playlists(page=0, page_size=0) second_page = reader.list_playlists(page=2, page_size=-9) self.assertEqual([3], [row["playlist_id"] for row in first_page]) self.assertEqual([2], [row["playlist_id"] for row in second_page]) def test_list_playlists_with_unknown_platform_does_not_fallback_to_all(self): self._insert_playlists( [ (1, "netease", "n-1", "Netease A", "desc", "https://img/a.jpg", 300, 10, 9), (2, "qq", "q-1", "QQ A", "desc", "https://img/b.jpg", 200, 10, 8), (3, "kuwo", "k-1", "Kuwo A", "desc", "https://img/c.jpg", 100, 10, 7), ] ) reader = CatalogReader(db_path=str(self._db_path)) rows = reader.list_playlists(page=1, page_size=10, platform="kugou") self.assertEqual([], rows) def test_get_playlist_returns_row_when_found_and_none_when_missing(self): self._insert_playlists( [ (42, "netease", "42", "Fav", "desc", "https://img/f.jpg", 88, 5, 3), ] ) reader = CatalogReader(db_path=str(self._db_path)) found = reader.get_playlist(42) missing = reader.get_playlist(404) self.assertIsNotNone(found) assert found is not None self.assertEqual(42, found["playlist_id"]) self.assertEqual("netease", found["platform"]) self.assertEqual(3, found["playable_song_count"]) self.assertIsNone(missing) def test_list_playlist_tracks_filters_non_active_and_keeps_position_order_with_pagination( self, ): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_playlist_tracks ( playlist_id integer not null, song_id integer not null, position integer not null ); """ ) conn.executemany( """ insert into catalog_tracks ( song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (1001, "qq", "q-1001", "Track 1001", "Singer A", "Album A", None, 100000, "{}"), (1002, "qq", "q-1002", "Track 1002", "Singer B", "Album B", None, 120000, "{}"), (1003, "qq", "q-1003", "Track 1003", "Singer C", "Album C", None, 140000, "{}"), (1004, "qq", "q-1004", "Track 1004", "Singer D", "Album D", None, 160000, "{}"), (1005, "qq", "q-1005", "Track 1005", "Singer E", "Album E", None, 180000, "{}"), ], ) conn.executemany( """ insert into catalog_playlist_tracks (playlist_id, song_id, position) values (?, ?, ?) """, [ (7, 1002, 1), (7, 1003, 2), (7, 1004, 3), (7, 1001, 4), (7, 1005, 5), ], ) conn.executemany( """ insert into catalog_track_files ( song_id, quality_label, ext, file_size_bytes, backend_type, backend_name, locator, public_url, status, is_primary ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (1001, "standard", "mp3", 101, "object_storage", "cdn", "1001.mp3", None, "active", 1), (1002, "standard", "mp3", 102, "object_storage", "cdn", "1002.mp3", None, "inactive", 1), (1003, "standard", "mp3", 103, "object_storage", "cdn", "1003.mp3", None, "active", 1), (1004, "standard", "mp3", 104, "object_storage", "cdn", "1004.mp3", None, "active", 1), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) full_rows = reader.list_playlist_tracks(playlist_id=7, page=1, page_size=10) first_page = reader.list_playlist_tracks(playlist_id=7, page=1, page_size=2) second_page = reader.list_playlist_tracks(playlist_id=7, page=2, page_size=2) third_page = reader.list_playlist_tracks(playlist_id=7, page=3, page_size=2) self.assertEqual([1003, 1004, 1001], [row["song_id"] for row in full_rows]) self.assertEqual([1003, 1004], [row["song_id"] for row in first_page]) self.assertEqual([1001], [row["song_id"] for row in second_page]) self.assertEqual([], third_page) def test_list_playlist_tracks_has_stable_secondary_sort_when_position_duplicates(self): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_playlist_tracks ( playlist_id integer not null, song_id integer not null, position integer not null ); """ ) conn.executemany( """ insert into catalog_tracks ( song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (1001, "qq", "q-1001", "Track 1001", "Singer A", "Album A", None, 100000, "{}"), (1002, "qq", "q-1002", "Track 1002", "Singer B", "Album B", None, 120000, "{}"), (1003, "qq", "q-1003", "Track 1003", "Singer C", "Album C", None, 140000, "{}"), ], ) conn.executemany( """ insert into catalog_playlist_tracks (playlist_id, song_id, position) values (?, ?, ?) """, [ (8, 1002, 1), (8, 1001, 1), (8, 1003, 2), ], ) conn.executemany( """ insert into catalog_track_files ( song_id, quality_label, ext, file_size_bytes, backend_type, backend_name, locator, public_url, status, is_primary ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (1001, "standard", "mp3", 101, "object_storage", "cdn", "1001.mp3", None, "active", 1), (1002, "standard", "mp3", 102, "object_storage", "cdn", "1002.mp3", None, "active", 1), (1003, "standard", "mp3", 103, "object_storage", "cdn", "1003.mp3", None, "active", 1), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) full_rows = reader.list_playlist_tracks(playlist_id=8, page=1, page_size=10) page1 = reader.list_playlist_tracks(playlist_id=8, page=1, page_size=1) page2 = reader.list_playlist_tracks(playlist_id=8, page=2, page_size=1) page3 = reader.list_playlist_tracks(playlist_id=8, page=3, page_size=1) self.assertEqual([1001, 1002, 1003], [row["song_id"] for row in full_rows]) self.assertEqual([1001], [row["song_id"] for row in page1]) self.assertEqual([1002], [row["song_id"] for row in page2]) self.assertEqual([1003], [row["song_id"] for row in page3]) def test_get_toplist_and_list_toplist_tracks_with_pagination(self): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_toplists ( toplist_id text primary key, platform text not null, name text not null, description text, cover_url text, play_count integer not null, song_count integer not null, playable_song_count integer not null, group_name text not null ); create table catalog_toplist_tracks ( toplist_id text not null, song_id integer not null, position integer not null ); """ ) conn.execute( """ insert into catalog_toplists ( toplist_id, platform, name, description, cover_url, play_count, song_count, playable_song_count, group_name ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( "kuwo_top_16", "kuwo", "Kuwo Hot Songs", "top songs", "https://img/top-16.jpg", 999, 3, 2, "Kuwo", ), ) conn.executemany( """ insert into catalog_tracks ( song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( 2001, "kuwo", "kw-2001", "Song 2001", "Singer A", "Album A", "https://img/2001.jpg", 210000, "{}", ), ( 2002, "kuwo", "kw-2002", "Song 2002", "Singer B", "Album B", "https://img/2002.jpg", 220000, "{}", ), ( 2003, "kuwo", "kw-2003", "Song 2003", "Singer C", "Album C", "https://img/2003.jpg", 200000, "{}", ), ], ) conn.executemany( """ insert into catalog_track_files ( song_id, quality_label, ext, file_size_bytes, backend_type, backend_name, locator, public_url, status, is_primary ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( 2001, "super", "flac", 100, "object_storage", "cdn", "song-2001.flac", "https://cdn/2001.flac", "active", 1, ), ( 2002, "super", "flac", 100, "object_storage", "cdn", "song-2002.flac", "https://cdn/2002.flac", "inactive", 1, ), ( 2003, "super", "flac", 100, "object_storage", "cdn", "song-2003.flac", "https://cdn/2003.flac", "active", 1, ), ], ) conn.executemany( """ insert into catalog_toplist_tracks (toplist_id, song_id, position) values (?, ?, ?) """, [ ("kuwo_top_16", 2002, 1), ("kuwo_top_16", 2001, 1), ("kuwo_top_16", 2003, 2), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) toplist_groups = reader.list_toplists() missing_toplist = reader.get_toplist("missing") toplist = reader.get_toplist("kuwo_top_16") first_page = reader.list_toplist_tracks("kuwo_top_16", page=0, page_size=0) second_page = reader.list_toplist_tracks("kuwo_top_16", page=2, page_size=1) missing_tracks = reader.list_toplist_tracks("missing", page=1, page_size=10) self.assertEqual(1, len(toplist_groups)) self.assertEqual("Kuwo", toplist_groups[0]["title"]) self.assertEqual(2, toplist_groups[0]["data"][0]["playable_song_count"]) self.assertIsNone(missing_toplist) self.assertIsNotNone(toplist) assert toplist is not None self.assertEqual("kuwo_top_16", toplist["toplist_id"]) self.assertEqual("kuwo", toplist["platform"]) self.assertEqual("Kuwo Hot Songs", toplist["name"]) self.assertEqual(2, toplist["playable_song_count"]) self.assertEqual([2001], [row["song_id"] for row in first_page]) self.assertEqual([2003], [row["song_id"] for row in second_page]) self.assertEqual([], missing_tracks) def test_list_playlists_closes_connection_when_query_raises(self): class FakeConnection: def __init__(self) -> None: self.closed = False def execute(self, *_args, **_kwargs): raise RuntimeError("query failed") def close(self) -> None: self.closed = True fake_conn = FakeConnection() original_connect = catalog_reader_module.connect_sqlite catalog_reader_module.connect_sqlite = lambda _db_path: fake_conn self.addCleanup( lambda: setattr(catalog_reader_module, "connect_sqlite", original_connect) ) reader = CatalogReader(db_path=str(self._db_path)) with self.assertRaises(RuntimeError): reader.list_playlists(page=1, page_size=10) self.assertTrue(fake_conn.closed) def test_search_tracks_prefers_name_match_and_requires_active_file(self): conn = sqlite3.connect(self._db_path) conn.executemany( """ insert into catalog_tracks ( song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( 1, "netease", "n1", "Moonlight", "Artist A", "A", "https://img/1.jpg", 210000, "{}", ), ( 2, "netease", "n2", "Moonlight Demo", "Artist B", "B", "https://img/2.jpg", 180000, "{}", ), ( 3, "qq", "q3", "Sunrise", "Moonlight Singer", "C", "https://img/3.jpg", 200000, "{}", ), ], ) conn.executemany( """ insert into catalog_track_files ( song_id, quality_label, ext, file_size_bytes, backend_type, backend_name, locator, public_url, status, is_primary ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( 1, "super", "flac", 100, "object_storage", "cdn", "song-1.flac", "https://cdn/1.flac", "active", 1, ), ( 2, "super", "flac", 101, "object_storage", "cdn", "song-2.flac", "https://cdn/2.flac", "inactive", 1, ), ( 3, "standard", "mp3", 99, "object_storage", "cdn", "song-3.mp3", "https://cdn/3.mp3", "active", 1, ), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) rows = reader.search_tracks(query="Moonlight", page=1, page_size=10) self.assertEqual([1, 3], [row["song_id"] for row in rows]) self.assertEqual("Moonlight", rows[0]["name"]) self.assertEqual("Moonlight Singer", rows[1]["singers"]) self.assertEqual( { "song_id", "name", "singers", "album", "cover_url", "duration_ms", "local_locator", }, set(rows[0].keys()), ) def test_search_tracks_blank_query_and_like_wildcard_escape(self): conn = sqlite3.connect(self._db_path) conn.executemany( """ insert into catalog_tracks ( song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( 11, "netease", "n11", "Moonlight", "Singer A", "A", "https://img/11.jpg", 210000, "{}", ), ( 12, "qq", "q12", "Sunrise", "Singer B", "B", "https://img/12.jpg", 200000, "{}", ), ( 13, "qq", "q13", "100% Love", "Singer C", "C", "https://img/13.jpg", 230000, "{}", ), ], ) conn.executemany( """ insert into catalog_track_files ( song_id, quality_label, ext, file_size_bytes, backend_type, backend_name, locator, public_url, status, is_primary ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( 11, "super", "flac", 100, "object_storage", "cdn", "song-11.flac", "https://cdn/11.flac", "active", 1, ), ( 12, "standard", "mp3", 90, "object_storage", "cdn", "song-12.mp3", "https://cdn/12.mp3", "active", 1, ), ( 13, "super", "flac", 110, "object_storage", "cdn", "song-13.flac", "https://cdn/13.flac", "active", 1, ), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) self.assertEqual([], reader.search_tracks(query=" ", page=1, page_size=10)) rows = reader.search_tracks(query="%", page=1, page_size=10) self.assertEqual([13], [row["song_id"] for row in rows]) def test_search_sheets_merges_playlists_and_toplists(self): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_toplists ( toplist_id text primary key, platform text not null, name text not null, description text, cover_url text, play_count integer not null, song_count integer not null, playable_song_count integer not null, group_name text not null ); """ ) self._insert_playlists( [ ( 1, "netease", "p-1", "Mix Daily", "playlist-desc", "https://img/p1.jpg", 300, 10, 6, ), ( 2, "qq", "p-2", "Hidden Mix", "playlist-desc", "https://img/p2.jpg", 200, 10, 0, ), ] ) conn.executemany( """ insert into catalog_toplists ( toplist_id, platform, name, description, cover_url, play_count, song_count, playable_song_count, group_name ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ("qq_top_mix", "qq", "Mix Rank", "top-desc", "https://img/top.jpg", 250, 8, 5, "QQ"), ("qq_top_blocked", "qq", "Mix Blocked", "top-desc", "https://img/top2.jpg", 260, 8, 0, "QQ"), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) rows = reader.search_sheets(query="Mix", page=1, page_size=10) self.assertEqual( [("playlist", "1"), ("toplist", "qq_top_mix")], [(row["item_type"], row["item_id"]) for row in rows], ) self.assertEqual([300, 250], [row["play_count"] for row in rows]) def test_search_sheets_blank_query_and_like_wildcard_escape(self): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_toplists ( toplist_id text primary key, platform text not null, name text not null, description text, cover_url text, play_count integer not null, song_count integer not null, playable_song_count integer not null, group_name text not null ); """ ) self._insert_playlists( [ ( 1, "qq", "p-1", "100% Mix", "playlist-desc", "https://img/p1.jpg", 20, 10, 2, ), ( 2, "qq", "p-2", "100_ Mix", "playlist-desc", "https://img/p2.jpg", 19, 10, 2, ), ( 3, "qq", "p-3", "100a Mix", "playlist-desc", "https://img/p3.jpg", 18, 10, 2, ), ] ) conn.executemany( """ insert into catalog_toplists ( toplist_id, platform, name, description, cover_url, play_count, song_count, playable_song_count, group_name ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( "top_percent", "qq", "Top 100% Mix", "top-desc", "https://img/top-percent.jpg", 17, 8, 2, "QQ", ), ( "top_under", "qq", "Top 100_ Mix", "top-desc", "https://img/top-under.jpg", 16, 8, 2, "QQ", ), ( "top_plain", "qq", "Top 100x Mix", "top-desc", "https://img/top-plain.jpg", 15, 8, 2, "QQ", ), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) self.assertEqual([], reader.search_sheets(query=" ", page=1, page_size=10)) percent_rows = reader.search_sheets(query="%", page=1, page_size=10) self.assertEqual( {("playlist", "1"), ("toplist", "top_percent")}, {(row["item_type"], row["item_id"]) for row in percent_rows}, ) underscore_rows = reader.search_sheets(query="_", page=1, page_size=10) self.assertEqual( {("playlist", "2"), ("toplist", "top_under")}, {(row["item_type"], row["item_id"]) for row in underscore_rows}, ) def test_search_sheets_ranking_tiebreakers_and_pagination(self): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_toplists ( toplist_id text primary key, platform text not null, name text not null, description text, cover_url text, play_count integer not null, song_count integer not null, playable_song_count integer not null, group_name text not null ); """ ) self._insert_playlists( [ (1, "netease", "p-1", "Mix", "playlist-desc", "https://img/p1.jpg", 500, 10, 6), (3, "qq", "p-3", "Mix", "playlist-desc", "https://img/p3.jpg", 500, 10, 6), (2, "qq", "p-2", "Mix Alpha", "playlist-desc", "https://img/p2.jpg", 700, 10, 6), (4, "qq", "p-4", "Alpha Mix", "playlist-desc", "https://img/p4.jpg", 900, 10, 6), ] ) conn.executemany( """ insert into catalog_toplists ( toplist_id, platform, name, description, cover_url, play_count, song_count, playable_song_count, group_name ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ ( "a_top_exact", "qq", "Mix", "top-desc", "https://img/top-a.jpg", 500, 8, 5, "QQ", ), ( "b_top_exact", "qq", "Mix", "top-desc", "https://img/top-b.jpg", 500, 8, 5, "QQ", ), ( "z_top_exact_low", "qq", "Mix", "top-desc", "https://img/top-z.jpg", 450, 8, 5, "QQ", ), ( "p_top_prefix", "qq", "Mix Beta", "top-desc", "https://img/top-p.jpg", 700, 8, 5, "QQ", ), ( "f_top_fuzzy", "qq", "Beta Mix", "top-desc", "https://img/top-f.jpg", 950, 8, 5, "QQ", ), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) page1 = reader.search_sheets(query="mix", page=1, page_size=4) page2 = reader.search_sheets(query="mix", page=2, page_size=4) self.assertEqual( [ ("playlist", "1"), ("playlist", "3"), ("toplist", "a_top_exact"), ("toplist", "b_top_exact"), ], [(row["item_type"], row["item_id"]) for row in page1], ) self.assertEqual([500, 500, 500, 500], [row["play_count"] for row in page1]) self.assertEqual( [ ("toplist", "z_top_exact_low"), ("playlist", "2"), ("toplist", "p_top_prefix"), ("toplist", "f_top_fuzzy"), ], [(row["item_type"], row["item_id"]) for row in page2], ) self.assertEqual([450, 700, 700, 950], [row["play_count"] for row in page2]) def test_search_artists_returns_empty_for_blank_query(self): reader = CatalogReader(db_path=str(self._db_path)) self.assertEqual([], reader.search_artists(query=" ", page=1, page_size=10)) def test_search_artists_applies_ranking_filter_and_pagination(self): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_artists ( artist_id integer primary key, artist_key text not null unique, platform text not null, remote_artist_id text, name text not null, normalized_name text not null, avatar_url text, description text, playable_song_count integer not null ); """ ) conn.executemany( """ insert into catalog_artists ( artist_id, artist_key, platform, remote_artist_id, name, normalized_name, avatar_url, description, playable_song_count ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (101, "netease:exact", "netease", "exact", "Singer", "singer", None, None, 1), (102, "netease:prefix-a", "netease", "prefix-a", "Singer Alpha", "singer alpha", None, None, 4), (103, "qq:prefix-b1", "qq", "prefix-b1", "Singer Beta", "singer beta", None, None, 4), (104, "qq:prefix-b2", "qq", "prefix-b2", "Singer Beta", "singer beta", None, None, 4), (109, "kuwo:prefix-o", "kuwo", "prefix-o", "Singer Omega", "singer omega", None, None, 2), (106, "netease:fuzzy-a", "netease", "fuzzy-a", "A Singer", "a singer", None, None, 10), (105, "netease:fuzzy-z", "netease", "fuzzy-z", "Z Singer", "z singer", None, None, 10), (108, "qq:fuzzy-m", "qq", "fuzzy-m", "M Singer", "m singer", None, None, 8), (110, "qq:filtered", "qq", "filtered", "Singer Zero", "singer zero", None, None, 0), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) page1 = reader.search_artists(query="Singer", page=1, page_size=4) page2 = reader.search_artists(query="Singer", page=2, page_size=4) self.assertEqual([101, 102, 103, 104], [row["artist_id"] for row in page1]) self.assertEqual([109, 106, 105, 108], [row["artist_id"] for row in page2]) self.assertEqual([2, 10, 10, 8], [row["playable_song_count"] for row in page2]) self.assertNotIn(110, [row["artist_id"] for row in page1 + page2]) def test_list_artist_tracks_filters_active_sorts_stably_and_paginates(self): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_artist_tracks ( artist_id integer not null, song_id integer not null, position integer not null ); """ ) conn.executemany( """ insert into catalog_tracks ( song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (201, "qq", "q-201", "Track 201", "Singer X", "Album X", None, 180000, "{}"), (202, "qq", "q-202", "Track 202", "Singer X", "Album X", None, 180000, "{}"), (203, "qq", "q-203", "Track 203", "Singer X", "Album X", None, 180000, "{}"), (204, "qq", "q-204", "Track 204", "Singer X", "Album X", None, 180000, "{}"), (205, "qq", "q-205", "Track 205", "Singer X", "Album X", None, 180000, "{}"), ], ) conn.executemany( """ insert into catalog_artist_tracks (artist_id, song_id, position) values (?, ?, ?) """, [ (88, 203, 1), (88, 201, 1), (88, 202, 2), (88, 204, 2), (88, 205, 3), ], ) conn.executemany( """ insert into catalog_track_files ( song_id, quality_label, ext, file_size_bytes, backend_type, backend_name, locator, public_url, status, is_primary ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (201, "standard", "mp3", 90, "object_storage", "cdn", "201.mp3", None, "active", 1), (202, "standard", "mp3", 90, "object_storage", "cdn", "202.mp3", None, "inactive", 1), (203, "standard", "mp3", 90, "object_storage", "cdn", "203.mp3", None, "active", 1), (204, "standard", "mp3", 90, "object_storage", "cdn", "204.mp3", None, "active", 1), (205, "standard", "mp3", 90, "object_storage", "cdn", "205.mp3", None, "active", 1), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) page1 = reader.list_artist_tracks(artist_id=88, page=1, page_size=2) page2 = reader.list_artist_tracks(artist_id=88, page=2, page_size=2) page3 = reader.list_artist_tracks(artist_id=88, page=3, page_size=2) full_rows = reader.list_artist_tracks(artist_id=88, page=1, page_size=10) self.assertEqual([201, 203, 204, 205], [row["song_id"] for row in full_rows]) self.assertEqual([201, 203], [row["song_id"] for row in page1]) self.assertEqual([204, 205], [row["song_id"] for row in page2]) self.assertEqual([], page3) def test_search_artists_get_artist_and_list_artist_tracks(self): conn = sqlite3.connect(self._db_path) conn.executescript( """ create table catalog_artists ( artist_id integer primary key, artist_key text not null unique, platform text not null, remote_artist_id text, name text not null, normalized_name text not null, avatar_url text, description text, playable_song_count integer not null ); create table catalog_artist_tracks ( artist_id integer not null, song_id integer not null, position integer not null ); """ ) conn.executemany( """ insert into catalog_artists ( artist_id, artist_key, platform, remote_artist_id, name, normalized_name, avatar_url, description, playable_song_count ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (1, "netease:artist-a", "netease", "artist-a", "Singer A", "singer a", "https://img/a.jpg", "desc-a", 1), (2, "qq:artist-b", "qq", "artist-b", "Singer B", "singer b", "https://img/b.jpg", "desc-b", 2), ], ) conn.executemany( """ insert into catalog_tracks ( song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (101, "netease", "n-101", "Alpha Song", "Singer A", "Album A", "https://img/song-a.jpg", 210000, "{}"), (102, "netease", "n-102", "Blocked Song", "Singer A", "Album A", "https://img/song-b.jpg", 180000, "{}"), (103, "qq", "q-103", "Bravo Song", "Singer B", "Album B", "https://img/song-c.jpg", 220000, "{}"), ], ) conn.executemany( """ insert into catalog_artist_tracks (artist_id, song_id, position) values (?, ?, ?) """, [(1, 101, 1), (1, 102, 2), (2, 103, 1)], ) conn.executemany( """ insert into catalog_track_files ( song_id, quality_label, ext, file_size_bytes, backend_type, backend_name, locator, public_url, status, is_primary ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (101, "super", "flac", 100, "object_storage", "cdn", "101.flac", "https://cdn/101.flac", "active", 1), (102, "standard", "mp3", 90, "object_storage", "cdn", "102.mp3", "https://cdn/102.mp3", "inactive", 1), (103, "super", "flac", 120, "object_storage", "cdn", "103.flac", "https://cdn/103.flac", "active", 1), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) artist_rows = reader.search_artists(query="Singer", page=1, page_size=10) artist = reader.get_artist(artist_id=1) tracks = reader.list_artist_tracks(artist_id=1, page=1, page_size=10) self.assertEqual([2, 1], [row["artist_id"] for row in artist_rows]) self.assertEqual([2, 1], [row["playable_song_count"] for row in artist_rows]) self.assertIsNotNone(artist) assert artist is not None self.assertEqual("netease", artist["platform"]) self.assertEqual("Singer A", artist["name"]) self.assertEqual([101], [row["song_id"] for row in tracks]) def test_count_playable_tracks_counts_distinct_active_song_ids(self): conn = sqlite3.connect(self._db_path) conn.executemany( """ insert into catalog_track_files ( song_id, quality_label, ext, file_size_bytes, backend_type, backend_name, locator, public_url, status, is_primary ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (1, "super", "flac", 100, "object_storage", "cdn", "song-1.flac", None, "active", 1), (1, "standard", "mp3", 80, "object_storage", "cdn", "song-1.mp3", None, "active", 0), (2, "standard", "mp3", 90, "object_storage", "cdn", "song-2.mp3", None, "inactive", 1), (3, "standard", "mp3", 90, "object_storage", "cdn", "song-3.mp3", None, "active", 1), ], ) conn.commit() conn.close() reader = CatalogReader(db_path=str(self._db_path)) self.assertEqual(2, reader.count_playable_tracks()) if __name__ == "__main__": unittest.main()