import sqlite3 import tempfile import unittest from pathlib import Path from unittest.mock import patch from fastapi.testclient import TestClient from music_server.app import create_app from tests.support import auth_headers class MfCatalogRouteTests(unittest.TestCase): def _prepare_playlist_toplist_catalog_db(self, db_path: Path) -> None: conn = sqlite3.connect(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_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_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 ); create table catalog_playlist_tracks ( playlist_id integer not null, song_id integer not null, position integer not null ); create table catalog_toplist_tracks ( toplist_id text not null, song_id integer not null, position integer not null ); 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.execute( """ insert into catalog_playlists ( playlist_id, platform, remote_playlist_id, name, description, cover_url, play_count, song_count, playable_song_count ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (1, "netease", "rid-1", "playlist-1", "desc", "https://img/1.jpg", 100, 2, 1), ) conn.execute( """ insert into catalog_toplists ( toplist_id, platform, name, description, cover_url, play_count, song_count, playable_song_count, group_name ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ("tl-1", "netease", "toplist-1", "desc", "https://img/top.jpg", 88, 2, 1, "official"), ) 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", "Playable Song", "Singer A", "Album A", "https://img/song1.jpg", 200000, "{}", ), ( 2, "netease", "n2", "Blocked Song", "Singer B", "Album B", "https://img/song2.jpg", 180000, "{}", ), ], ) 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, "standard", "mp3", 90, "object_storage", "cdn", "song-2.mp3", "https://cdn/2.mp3", "inactive", 1, ), ], ) conn.executemany( """ insert into catalog_playlist_tracks (playlist_id, song_id, position) values (?, ?, ?) """, [(1, 1, 1), (1, 2, 2)], ) conn.executemany( """ insert into catalog_toplist_tracks (toplist_id, song_id, position) values (?, ?, ?) """, [("tl-1", 1, 1), ("tl-1", 2, 2)], ) conn.execute( """ 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/artist-a.jpg", "artist-desc", 1, ), ) conn.execute( """ insert into catalog_artist_tracks (artist_id, song_id, position) values (?, ?, ?) """, (1, 1, 1), ) conn.commit() conn.close() def test_recommend_tags_returns_musicfree_shape(self): with tempfile.TemporaryDirectory() as tmpdir: player_db_path = Path(tmpdir) / "player.db" with patch.dict("os.environ", {"PLAYER_DB_PATH": str(player_db_path)}, clear=False): client = TestClient(create_app()) response = client.get( "/mf/v1/recommend/tags", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertIn("pinned", payload) self.assertIn("data", payload) self.assertEqual(["all", "netease", "qq", "kuwo"], [item["id"] for item in payload["pinned"]]) self.assertEqual( ["playlist_square", "toplist"], [item["id"] for item in payload["data"][0]["data"]], ) def test_recommend_routes_requires_token_when_missing(self): client = TestClient(create_app()) response = client.get("/mf/v1/recommend/tags") self.assertEqual(401, response.status_code) def test_recommend_routes_requires_token_when_wrong(self): with tempfile.TemporaryDirectory() as tmpdir: player_db_path = Path(tmpdir) / "player.db" with patch.dict("os.environ", {"PLAYER_DB_PATH": str(player_db_path)}, clear=False): client = TestClient(create_app()) response = client.get( "/mf/v1/recommend/tags", headers=auth_headers(player_db_path, token="wrong-token"), ) self.assertEqual(401, response.status_code) def test_recommend_routes_allow_anonymous_when_auth_disabled(self): with tempfile.TemporaryDirectory() as tmpdir: player_db_path = Path(tmpdir) / "player.db" with patch.dict( "os.environ", { "PLAYER_DB_PATH": str(player_db_path), "MUSIC_SERVER_DISABLE_AUTH": "1", }, clear=False, ): client = TestClient(create_app()) response = client.get("/mf/v1/recommend/tags") self.assertEqual(200, response.status_code) def test_recommend_sheets_returns_musicfree_shape(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" conn = sqlite3.connect(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 ); """ ) rows = [ ( 1, "netease", "rid-1", "测试歌单", "desc", "https://img/1.jpg", 999, 9, 5, ) ] for idx in range(2, 21): rows.append( ( idx, "netease", f"rid-{idx}", f"playlist-{idx}", "desc", f"https://img/{idx}.jpg", 200 - idx, 5, 5, ) ) 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() with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), }, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/recommend/sheets?page=1&page_size=20", headers=auth_headers(player_db_path), ) response_large_page = client.get( "/mf/v1/recommend/sheets?page=1&page_size=21", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertFalse(payload["isEnd"]) self.assertEqual("catalogsync:playlist:1", payload["data"][0]["id"]) self.assertEqual("测试歌单", payload["data"][0]["title"]) self.assertEqual(9, payload["data"][0]["worksNum"]) self.assertEqual(5, payload["data"][0]["playableSongCount"]) self.assertEqual(999, payload["data"][0]["play_count"]) self.assertNotIn("playCount", payload["data"][0]) self.assertEqual(200, response_large_page.status_code) self.assertTrue(response_large_page.json()["isEnd"]) def test_recommend_sheets_filters_by_platform_tag(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" conn = sqlite3.connect(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 ); """ ) conn.executemany( """ insert into catalog_playlists ( playlist_id, platform, remote_playlist_id, name, description, cover_url, play_count, song_count, playable_song_count ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, [ (1, "netease", "n-1", "Netease List", "desc", "https://img/1.jpg", 300, 10, 10), (2, "qq", "q-1", "QQ List", "desc", "https://img/2.jpg", 200, 10, 10), (3, "kuwo", "k-1", "Kuwo List", "desc", "https://img/3.jpg", 100, 10, 10), ], ) conn.commit() conn.close() with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), }, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/recommend/sheets?tag=qq&page=1&page_size=20", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertTrue(payload["isEnd"]) self.assertEqual(["catalogsync:playlist:2"], [item["id"] for item in payload["data"]]) self.assertEqual(["QQ List"], [item["title"] for item in payload["data"]]) def test_recommend_sheets_returns_toplists_when_tag_toplist(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" self._prepare_playlist_toplist_catalog_db(db_path) conn = sqlite3.connect(db_path) conn.execute( """ insert into catalog_toplists ( toplist_id, platform, name, description, cover_url, play_count, song_count, playable_song_count, group_name ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ("tl-2", "netease", "toplist-2", "desc", "https://img/top2.jpg", 77, 1, 1, "official"), ) conn.commit() conn.close() with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), }, clear=False, ): client = TestClient(create_app()) response_page_1 = client.get( "/mf/v1/recommend/sheets?tag=toplist&page=1&page_size=1", headers=auth_headers(player_db_path), ) response_page_2 = client.get( "/mf/v1/recommend/sheets?tag=toplist&page=2&page_size=1", headers=auth_headers(player_db_path), ) self.assertEqual(200, response_page_1.status_code) self.assertEqual(200, response_page_2.status_code) payload_page_1 = response_page_1.json() payload_page_2 = response_page_2.json() self.assertEqual(["catalogsync:toplist:tl-1"], [item["id"] for item in payload_page_1["data"]]) self.assertTrue(payload_page_1["data"][0]["id"].startswith("catalogsync:toplist:")) self.assertFalse(payload_page_1["isEnd"]) self.assertEqual(["catalogsync:toplist:tl-2"], [item["id"] for item in payload_page_2["data"]]) self.assertTrue(payload_page_2["data"][0]["id"].startswith("catalogsync:toplist:")) self.assertTrue(payload_page_2["isEnd"]) def test_search_songs_returns_musicfree_shape(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" self._prepare_playlist_toplist_catalog_db(db_path) with patch.dict( "os.environ", {"CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path)}, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/search/songs?q=Playable&page=1&page_size=20", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertTrue(payload["isEnd"]) self.assertEqual(["catalogsync:song:1"], [item["id"] for item in payload["data"]]) def test_search_songs_includes_raw_lrc_when_local_lyrics_exist(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" library_root = Path(tmpdir) / "library" library_root.mkdir() self._prepare_playlist_toplist_catalog_db(db_path) conn = sqlite3.connect(db_path) conn.execute( """ 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, "standard", "mp3", 80, "local_fs", "library", "Singer A/Playable Song.mp3", None, "active", 0, ), ) conn.commit() conn.close() song_dir = library_root / "Singer A" song_dir.mkdir() (song_dir / "Playable Song.mp3").write_bytes(b"audio") (song_dir / "Playable Song.lrc").write_text("[00:00.00]hello lyric\n", encoding="utf-8") with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), "LOCAL_LIBRARY_ROOT": str(library_root), }, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/search/songs?q=Playable&page=1&page_size=20", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertEqual("[00:00.00]hello lyric\n", payload["data"][0]["rawLrc"]) def test_song_lyric_route_returns_raw_lrc_when_local_lyrics_exist(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" library_root = Path(tmpdir) / "library" library_root.mkdir() self._prepare_playlist_toplist_catalog_db(db_path) conn = sqlite3.connect(db_path) conn.execute( """ 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, "standard", "mp3", 80, "local_fs", "library", "Singer A/Playable Song.mp3", None, "active", 0, ), ) conn.commit() conn.close() song_dir = library_root / "Singer A" song_dir.mkdir() (song_dir / "Playable Song.mp3").write_bytes(b"audio") (song_dir / "Playable Song.lrc").write_text("[00:00.00]hello lyric\n", encoding="utf-8") with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), "LOCAL_LIBRARY_ROOT": str(library_root), }, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/songs/1/lyric", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertEqual("[00:00.00]hello lyric\n", payload["rawLrc"]) self.assertEqual("[00:00.00]hello lyric\n", payload["lyric"]) def test_playlist_tracks_omit_raw_lrc_when_local_library_root_missing(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" self._prepare_playlist_toplist_catalog_db(db_path) conn = sqlite3.connect(db_path) conn.execute( """ 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, "standard", "mp3", 80, "local_fs", "library", "Singer A/Playable Song.mp3", None, "active", 0, ), ) conn.commit() conn.close() with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), }, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/playlists/1/tracks?page=1&page_size=20", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertNotIn("rawLrc", payload["musicList"][0]) def test_search_artists_and_artist_detail_routes_return_musicfree_shape(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" self._prepare_playlist_toplist_catalog_db(db_path) with patch.dict( "os.environ", {"CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path)}, clear=False, ): client = TestClient(create_app()) search_response = client.get( "/mf/v1/search/artists?q=Singer&page=1&page_size=20", headers=auth_headers(player_db_path), ) detail_response = client.get( "/mf/v1/artists/1", headers=auth_headers(player_db_path), ) tracks_response = client.get( "/mf/v1/artists/1/tracks?page=1&page_size=20", headers=auth_headers(player_db_path), ) self.assertEqual(200, search_response.status_code) self.assertEqual(200, detail_response.status_code) self.assertEqual(200, tracks_response.status_code) self.assertEqual("catalogsync:artist:1", search_response.json()["data"][0]["id"]) self.assertEqual(["music"], search_response.json()["data"][0]["supportedArtistTabs"]) self.assertEqual("Singer A", detail_response.json()["name"]) self.assertEqual("catalogsync:song:1", tracks_response.json()["musicList"][0]["id"]) def test_search_sheets_returns_playlists_and_toplists(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" self._prepare_playlist_toplist_catalog_db(db_path) with patch.dict( "os.environ", {"CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path)}, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/search/sheets?q=1&page=1&page_size=20", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertEqual( ["catalogsync:playlist:1", "catalogsync:toplist:tl-1"], [item["id"] for item in payload["data"]], ) def test_playlist_tracks_only_returns_playable_rows(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" self._prepare_playlist_toplist_catalog_db(db_path) with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), }, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/playlists/1/tracks?page=1&page_size=20", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertEqual(1, len(payload["musicList"])) self.assertEqual("catalogsync:song:1", payload["musicList"][0]["id"]) self.assertEqual("Playable Song", payload["musicList"][0]["title"]) def test_toplists_returns_playable_song_count(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" self._prepare_playlist_toplist_catalog_db(db_path) with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), }, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/toplists", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertEqual(1, len(payload)) self.assertEqual(1, len(payload[0]["data"])) toplist = payload[0]["data"][0] self.assertEqual("catalogsync:toplist:tl-1", toplist["id"]) self.assertEqual(2, toplist["worksNum"]) self.assertEqual(1, toplist["playableSongCount"]) def test_toplist_tracks_only_returns_playable_rows(self): with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "catalog_read.db" player_db_path = Path(tmpdir) / "player.db" self._prepare_playlist_toplist_catalog_db(db_path) with patch.dict( "os.environ", { "CATALOG_DB_PATH": str(db_path), "PLAYER_DB_PATH": str(player_db_path), }, clear=False, ): client = TestClient(create_app()) response = client.get( "/mf/v1/toplists/tl-1/tracks?page=1&page_size=20", headers=auth_headers(player_db_path), ) self.assertEqual(200, response.status_code) payload = response.json() self.assertEqual(1, len(payload["musicList"])) self.assertEqual("catalogsync:song:1", payload["musicList"][0]["id"]) self.assertEqual("Playable Song", payload["musicList"][0]["title"]) if __name__ == "__main__": unittest.main()