xiaoming/musicdl-catalog-sync-suite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
musicdl-catalog-sync-suite/Music_Server/tests/test_cache_service.py
T

522 lines
21 KiB
Python
Raw Permalink Blame History

import sqlite3
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from music_server.services.cache_service import CacheService
class CacheServiceTests(unittest.TestCase):
def _prepare_catalog_db(self, db_path: Path) -> None:
conn = sqlite3.connect(db_path)
conn.execute(
"""
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
)
"""
)
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
)
"""
)
track_rows = [
(1001, "kuwo", "remote-1001", "Song 1001", "Singer 1001", "Album 1001", None, None, None),
(1002, "kuwo", "remote-1002", "Song 1002", "Singer 1002", "Album 1002", None, None, None),
(1003, "kuwo", "remote-1003", "Song 1003", "Singer 1003", "Album 1003", None, None, None),
(1004, "kuwo", "remote-1004", "Song 1004", "Singer 1004", "Album 1004", None, None, None),
]
conn.executemany(
"""
insert into catalog_tracks (
song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json
) values (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
track_rows,
)
rows = [
(1001, "standard", "mp3", 1000, "local_fs", "library", "std/1001.mp3", None, "active", 0),
(1001, "super", "flac", 2000, "local_fs", "library", "super/1001.flac", None, "active", 1),
(1002, "standard", "mp3", 1000, "local_fs", "library", "std/1002.mp3", None, "active", 1),
(1003, "standard", "mp3", 1000, "object_storage", "main", "obj/1003.mp3", "https://origin.example/1003.mp3", "active", 1),
(1004, "standard", "mp3", 1000, "object_storage", "main", "obj/1004.mp3", "https://origin.example/1004.mp3", "active", 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
rows,
)
conn.commit()
conn.close()
def test_record_stream_play_counts_each_stream_token_once(self):
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
)
first = service.record_stream_play(
song_id=1001,
stream_token="stream-a",
played_at="2026-04-23T10:00:00+00:00",
)
second = service.record_stream_play(
song_id=1001,
stream_token="stream-a",
played_at="2026-04-23T10:01:00+00:00",
)
summary = service.get_heat_summary(song_id=1001)
self.assertTrue(first)
self.assertFalse(second)
self.assertEqual(1, summary["play_count_total"])
self.assertEqual(1, summary["play_count_30d"])
self.assertEqual("2026-04-23T10:00:00+00:00", summary["last_played_at"])
def test_record_stream_play_is_noop_when_cache_relay_disabled(self):
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
cache_relay_enabled=False,
)
recorded = service.record_stream_play(
song_id=1001,
stream_token="stream-disabled",
played_at="2026-04-23T10:00:00+00:00",
)
summary = service.get_heat_summary(song_id=1001)
self.assertFalse(recorded)
self.assertEqual(0, summary["play_count_total"])
self.assertEqual(0, summary["play_count_30d"])
self.assertIsNone(summary["last_played_at"])
def test_reconcile_assigns_strict_top_n_targets_and_best_quality_source(self):
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
)
target_a = service.create_cache_target(
name="tier-a",
kind="sftp",
order_index=1,
capacity_songs=2,
public_base_url="https://cache-a.example",
path_prefix="music/a",
enabled=True,
secrets={"host": "127.0.0.1", "username": "tester"},
)
target_b = service.create_cache_target(
name="tier-b",
kind="s3",
order_index=2,
capacity_songs=1,
public_base_url="https://cache-b.example",
path_prefix="music/b",
enabled=True,
secrets={"bucket": "music", "region": "test"},
)
service.upsert_heat_summary(
song_id=1001,
play_count_total=50,
play_count_30d=50,
last_played_at="2026-04-23T12:00:00+00:00",
)
service.upsert_heat_summary(
song_id=1002,
play_count_total=40,
play_count_30d=40,
last_played_at="2026-04-23T11:00:00+00:00",
)
service.upsert_heat_summary(
song_id=1003,
play_count_total=30,
play_count_30d=30,
last_played_at="2026-04-23T10:00:00+00:00",
)
service.upsert_heat_summary(
song_id=1004,
play_count_total=20,
play_count_30d=20,
last_played_at="2026-04-23T09:00:00+00:00",
)
result = service.reconcile_cache_assignments(now="2026-04-23T13:00:00+00:00")
objects = service.list_cache_objects()
tasks = service.list_transfer_tasks()
self.assertEqual(3, result["desired_song_count"])
self.assertEqual(3, len(objects))
self.assertEqual(3, len(tasks))
self.assertEqual(
[
(1001, target_a["id"], "super", "super/1001.flac", "pending_upload"),
(1002, target_a["id"], "standard", "std/1002.mp3", "pending_upload"),
(1003, target_b["id"], "standard", "obj/1003.mp3", "pending_upload"),
],
[
(
item["song_id"],
item["target_id"],
item["quality_label"],
item["source_locator"],
item["status"],
)
for item in objects
],
)
def test_reconcile_marks_out_of_range_items_evictable_and_only_deletes_when_over_capacity(self):
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
)
target = service.create_cache_target(
name="tier-a",
kind="sftp",
order_index=1,
capacity_songs=1,
public_base_url="https://cache-a.example",
path_prefix="music",
enabled=True,
secrets={"host": "127.0.0.1", "username": "tester"},
)
service.upsert_cache_object(
song_id=1002,
target_id=target["id"],
quality_label="standard",
source_locator="std/1002.mp3",
remote_key="music/1002.mp3",
public_url="https://cache-a.example/music/1002.mp3",
status="active",
last_rank=1,
uploaded_at="2026-04-20T00:00:00+00:00",
last_verified_at="2026-04-20T00:00:00+00:00",
evictable=False,
)
service.upsert_cache_object(
song_id=1004,
target_id=target["id"],
quality_label="standard",
source_locator="obj/1004.mp3",
remote_key="music/1004.mp3",
public_url="https://cache-a.example/music/1004.mp3",
status="active",
last_rank=2,
uploaded_at="2026-04-20T00:00:00+00:00",
last_verified_at="2026-04-20T00:00:00+00:00",
evictable=False,
)
service.upsert_heat_summary(
song_id=1001,
play_count_total=10,
play_count_30d=10,
last_played_at="2026-04-23T12:00:00+00:00",
)
service.reconcile_cache_assignments(now="2026-04-23T13:00:00+00:00")
objects = service.list_cache_objects(target_id=target["id"])
tasks = service.list_transfer_tasks(task_kind="delete")
self.assertEqual(
[(1001, "pending_upload", False), (1002, "evictable", True)],
[(item["song_id"], item["status"], bool(item["evictable"])) for item in objects],
)
self.assertEqual([1004], [item["song_id"] for item in tasks])
def test_process_transfer_tasks_marks_uploaded_objects_active(self):
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
library_root = Path(tmpdir) / "library"
local_file = library_root / "super" / "1001.flac"
local_file.parent.mkdir(parents=True, exist_ok=True)
local_file.write_bytes(b"audio-data")
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
local_library_root=str(library_root),
)
service.create_cache_target(
name="tier-a",
kind="sftp",
order_index=1,
capacity_songs=1,
public_base_url="https://cache-a.example",
path_prefix="music",
enabled=True,
secrets={"host": "127.0.0.1", "username": "tester"},
)
service.upsert_heat_summary(
song_id=1001,
play_count_total=99,
play_count_30d=99,
last_played_at="2026-04-23T12:00:00+00:00",
)
service.reconcile_cache_assignments(now="2026-04-23T13:00:00+00:00")
with patch("music_server.services.cache_service.SFTPCacheTargetUploader.upload_file") as upload_file:
result = service.process_transfer_tasks(now="2026-04-23T13:05:00+00:00")
objects = service.list_cache_objects()
tasks = service.list_transfer_tasks()
self.assertEqual(1, result["uploaded"])
self.assertEqual("active", objects[0]["status"])
self.assertEqual("success", tasks[0]["status"])
upload_file.assert_called_once()
def test_reconcile_is_noop_when_cache_relay_disabled(self):
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
cache_relay_enabled=False,
)
service.create_cache_target(
name="tier-a",
kind="sftp",
order_index=1,
capacity_songs=2,
public_base_url="https://cache-a.example",
path_prefix="music/a",
enabled=True,
secrets={"host": "127.0.0.1", "username": "tester"},
)
service.upsert_heat_summary(
song_id=1001,
play_count_total=50,
play_count_30d=50,
last_played_at="2026-04-23T12:00:00+00:00",
)
result = service.reconcile_cache_assignments(now="2026-04-23T13:00:00+00:00")
self.assertEqual(0, result["desired_song_count"])
self.assertEqual(0, result["created_upload_tasks"])
self.assertEqual(0, result["created_delete_tasks"])
self.assertEqual([], service.list_cache_objects())
self.assertEqual([], service.list_transfer_tasks())
def test_process_transfer_tasks_is_noop_when_cache_relay_disabled(self):
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
cache_relay_enabled=False,
)
target = service.create_cache_target(
name="tier-a",
kind="sftp",
order_index=1,
capacity_songs=1,
public_base_url="https://cache-a.example",
path_prefix="music",
enabled=True,
secrets={"host": "127.0.0.1", "username": "tester"},
)
service.upsert_cache_object(
song_id=1001,
target_id=target["id"],
quality_label="super",
source_locator="super/1001.flac",
remote_key="music/1001_super.flac",
public_url="https://cache-a.example/music/1001_super.flac",
status="pending_upload",
last_rank=1,
uploaded_at=None,
last_verified_at=None,
evictable=False,
)
conn = sqlite3.connect(player_db_path)
conn.execute(
"""
insert into cache_transfer_tasks (
song_id, target_id, task_kind, quality_label, source_locator, remote_key,
public_url, status, run_id, created_at, updated_at
) values (?, ?, 'upload', ?, ?, ?, ?, 'pending', null, ?, ?)
""",
(
1001,
target["id"],
"super",
"super/1001.flac",
"music/1001_super.flac",
"https://cache-a.example/music/1001_super.flac",
"2026-04-23T13:00:00+00:00",
"2026-04-23T13:00:00+00:00",
),
)
conn.commit()
conn.close()
result = service.process_transfer_tasks(now="2026-04-23T13:05:00+00:00")
tasks = service.list_transfer_tasks()
objects = service.list_cache_objects()
self.assertEqual({"uploaded": 0, "deleted": 0, "failed": 0}, result)
self.assertEqual("pending", tasks[0]["status"])
self.assertEqual("pending_upload", objects[0]["status"])
def test_list_hot_songs_includes_name_and_prefers_cache_public_url(self):
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
)
target = service.create_cache_target(
name="tier-a",
kind="s3",
order_index=1,
capacity_songs=10,
public_base_url="https://cache-a.example",
path_prefix="music",
enabled=True,
secrets={"bucket": "music-cache", "access_key_id": "AKIA", "secret_access_key": "SECRET"},
)
service.upsert_heat_summary(
song_id=1001,
play_count_total=20,
play_count_30d=20,
last_played_at="2026-04-23T12:00:00+00:00",
)
service.upsert_heat_summary(
song_id=1003,
play_count_total=10,
play_count_30d=10,
last_played_at="2026-04-23T11:00:00+00:00",
)
service.upsert_cache_object(
song_id=1001,
target_id=target["id"],
quality_label="super",
source_locator="super/1001.flac",
remote_key="music/1001_super.flac",
public_url="https://cache-a.example/music/1001_super.flac",
status="active",
last_rank=1,
uploaded_at="2026-04-23T12:05:00+00:00",
last_verified_at="2026-04-23T12:05:00+00:00",
evictable=False,
)
hot_songs = service.list_hot_songs(limit=10)
self.assertEqual(
[
(1001, "Song 1001", "https://cache-a.example/music/1001_super.flac"),
(1003, "Song 1003", "https://origin.example/1003.mp3"),
],
[(item["song_id"], item["name"], item["external_url"]) for item in hot_songs],
)
@patch("music_server.services.cache_service.SFTPCacheTargetUploader")
def test_test_target_connection_payload_passes_sftp_remote_root(self, uploader_class):
uploader = uploader_class.return_value
with tempfile.TemporaryDirectory() as tmpdir:
catalog_db_path = Path(tmpdir) / "catalog_read.db"
player_db_path = Path(tmpdir) / "player.db"
self._prepare_catalog_db(catalog_db_path)
service = CacheService(
player_db_path=str(player_db_path),
catalog_db_path=str(catalog_db_path),
secret_encryption_key="test-secret",
)
result = service.test_target_connection_payload(
kind="sftp",
secrets={
"host": "64.83.43.123",
"port": 22,
"username": "root",
"password": "secret",
"remote_root": "/srv/music_server_cache",
},
)
uploader_class.assert_called_once_with(
host="64.83.43.123",
port=22,
username="root",
password="secret",
private_key=None,
timeout_seconds=10,
remote_root="/srv/music_server_cache",
)
uploader.test_connection.assert_called_once_with()
self.assertEqual("sftp", result["kind"])
self.assertTrue(result["ok"])
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink