xiaoming/musicdl-catalog-sync-suite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: sync catalog, server, and MusicFree updates

Browse Source
This commit is contained in:
unknown
2026-07-16 18:43:22 +08:00
parent 069af30dba
commit e9bb3df906
58 changed files with 4409 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Music_Server/tests/test_mf_media_routes.py
+201
View File
@@ -1,5 +1,6 @@
import sqlite3
import tempfile
import time
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -156,6 +157,42 @@ class MfMediaRouteTests(unittest.TestCase):
self.assertIn("/mf/v1/media/stream/", payload["stream"]["url"])
self.assertTrue(payload["stream"]["url"].endswith(".flac"))
def test_media_resolve_issues_longer_lived_stream_token_for_background_queueing(self):
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(db_path)
with patch.dict(
"os.environ",
{
"CATALOG_DB_PATH": str(db_path),
"PLAYER_DB_PATH": str(player_db_path),
"PUBLIC_MUSIC_ACCESS_TOKEN": "dev-token",
},
clear=False,
):
client = TestClient(create_app())
resolve_response = client.post(
"/mf/v1/media/resolve",
headers=auth_headers(player_db_path),
json={"song_id": "catalogsync:song:3476", "quality": "super"},
)
self.assertEqual(200, resolve_response.status_code)
stream_url = resolve_response.json()["stream"]["url"]
token = stream_url.rsplit("/", 1)[-1].split(".", 1)[0]
from music_server.services.stream_tokens import parse_stream_token
parsed = parse_stream_token(secret="dev-token", token=token)
remaining = parsed["expires_at"] - int(time.time())
self.assertGreaterEqual(
remaining,
1800,
"stream token ttl should be long enough for background playback queueing",
)
def test_media_stream_redirects_to_public_url(self):
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "catalog_read.db"
@@ -353,6 +390,170 @@ class MfMediaRouteTests(unittest.TestCase):
stream_response.headers.get("location"),
)
def test_media_stream_falls_back_to_current_active_source_when_token_locator_is_stale(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.execute(
"""
create table catalog_track_files (
song_id integer not null,
quality_label text not null,
ext text not null,
file_size_bytes integer not null,
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.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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
3476,
"super",
"flac",
42345678,
"object_storage",
"main-s3",
"music/netease/new.flac",
"https://cdn.example/new.flac",
"active",
1,
),
)
conn.commit()
conn.close()
with patch.dict(
"os.environ",
{
"CATALOG_DB_PATH": str(db_path),
"PLAYER_DB_PATH": str(player_db_path),
"PUBLIC_MUSIC_ACCESS_TOKEN": "dev-token",
},
clear=False,
):
client = TestClient(create_app())
from music_server.services.stream_tokens import create_stream_token
stale_token = create_stream_token(
secret="dev-token",
song_id=3476,
locator="music/netease/old.flac",
ttl_seconds=3600,
)
stream_response = client.get(
f"/mf/v1/media/stream/{stale_token}.flac",
follow_redirects=False,
)
self.assertEqual(307, stream_response.status_code)
self.assertEqual(
"https://cdn.example/new.flac",
stream_response.headers.get("location"),
)
def test_media_stream_falls_back_to_next_origin_source_when_selected_public_url_is_unreachable(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.execute(
"""
create table catalog_track_files (
song_id integer not null,
quality_label text not null,
ext text not null,
file_size_bytes integer not null,
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.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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
(
3476,
"super",
"flac",
42345678,
"object_storage",
"main-s3",
"music/netease/first.flac",
"https://cdn.example/first.flac",
"active",
1,
),
(
3476,
"super",
"flac",
42345678,
"object_storage",
"backup-s3",
"music/netease/second.flac",
"https://cdn.example/second.flac",
"active",
0,
),
],
)
conn.commit()
conn.close()
with patch.dict(
"os.environ",
{
"CATALOG_DB_PATH": str(db_path),
"PLAYER_DB_PATH": str(player_db_path),
"PUBLIC_MUSIC_ACCESS_TOKEN": "dev-token",
},
clear=False,
):
client = TestClient(create_app())
resolve_response = client.post(
"/mf/v1/media/resolve",
headers=auth_headers(player_db_path),
json={"song_id": "catalogsync:song:3476", "quality": "super"},
)
self.assertEqual(200, resolve_response.status_code)
with patch(
"music_server.routes.mf_media._is_origin_stream_reachable",
side_effect=lambda url: url.endswith("second.flac"),
):
stream_response = client.get(
resolve_response.json()["stream"]["url"],
follow_redirects=False,
)
self.assertEqual(307, stream_response.status_code)
self.assertEqual(
"https://cdn.example/second.flac",
stream_response.headers.get("location"),
)
def test_media_stream_falls_back_when_cached_public_url_is_unreachable(self):
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "catalog_read.db"