import hashlib 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 music_server.services.cache_service import CacheService class AdminCacheRouteTests(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, 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_tracks ( song_id, platform, remote_song_id, name, singers, album, cover_url, duration_ms, metadata_json ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( 2001, "kuwo", "remote-2001", "Song 2001", "Singer 2001", "Album 2001", None, None, None, ), ) 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( 2001, "super", "flac", 1024, "object_storage", "origin", "songs/2001.flac", "https://origin.example/2001.flac", "active", 1, ), ) conn.commit() conn.close() def _admin_env(self, player_db_path: Path, catalog_db_path: Path) -> dict[str, str]: return { "PLAYER_DB_PATH": str(player_db_path), "CATALOG_DB_PATH": str(catalog_db_path), "MUSIC_SERVER_ADMIN_USERNAME": "admin", "MUSIC_SERVER_ADMIN_PASSWORD_HASH": f"sha256${hashlib.sha256('secret123'.encode('utf-8')).hexdigest()}", "MUSIC_SERVER_SECRET_ENCRYPTION_KEY": "test-secret", } def test_admin_cache_api_requires_login(self): with tempfile.TemporaryDirectory() as tmpdir: player_db_path = Path(tmpdir) / "player.db" catalog_db_path = Path(tmpdir) / "catalog_read.db" self._prepare_catalog_db(catalog_db_path) with patch.dict("os.environ", self._admin_env(player_db_path, catalog_db_path), clear=False): client = TestClient(create_app()) response = client.get("/admin/api/cache/overview") self.assertEqual(401, response.status_code) def test_admin_login_then_manage_targets_and_reconcile(self): with tempfile.TemporaryDirectory() as tmpdir: player_db_path = Path(tmpdir) / "player.db" catalog_db_path = Path(tmpdir) / "catalog_read.db" self._prepare_catalog_db(catalog_db_path) with patch.dict("os.environ", self._admin_env(player_db_path, catalog_db_path), clear=False): service = CacheService( player_db_path=str(player_db_path), catalog_db_path=str(catalog_db_path), secret_encryption_key="test-secret", ) service.upsert_heat_summary( song_id=2001, play_count_total=12, play_count_30d=12, last_played_at="2026-04-23T12:00:00+00:00", ) client = TestClient(create_app()) login = client.post( "/admin/session/login", data={"username": "admin", "password": "secret123"}, follow_redirects=False, ) self.assertIn(login.status_code, {200, 204, 303}) create_target = client.post( "/admin/api/cache/targets", json={ "name": "tier-a", "kind": "s3", "order_index": 1, "capacity_songs": 10, "public_base_url": "https://cache.example", "path_prefix": "music", "enabled": True, "secrets": { "bucket": "music-cache", "region": "ap-shanghai", "access_key_id": "AKIA", "secret_access_key": "SECRET", }, }, ) overview = client.get("/admin/api/cache/overview") targets = client.get("/admin/api/cache/targets") hot_songs = client.get("/admin/api/cache/hot-songs") reconcile = client.post("/admin/api/cache/reconcile") html_page = client.get("/admin/cache") self.assertEqual(200, create_target.status_code) self.assertEqual(200, overview.status_code) self.assertEqual(200, targets.status_code) self.assertEqual(200, hot_songs.status_code) self.assertEqual(200, reconcile.status_code) self.assertEqual(200, html_page.status_code) target_payload = targets.json()["items"][0] self.assertNotIn("secrets", target_payload) self.assertTrue(target_payload["has_secrets"]) self.assertEqual( ["access_key_id", "bucket", "region", "secret_access_key"], target_payload["secret_fields"], ) self.assertEqual(2001, hot_songs.json()["items"][0]["song_id"]) self.assertEqual("Song 2001", hot_songs.json()["items"][0]["name"]) self.assertEqual("https://origin.example/2001.flac", hot_songs.json()["items"][0]["external_url"]) self.assertEqual(1, reconcile.json()["created_upload_tasks"]) self.assertIn("Cache Targets", html_page.text) self.assertIn('id="target-form"', html_page.text) self.assertIn('id="target-kind"', html_page.text) self.assertIn('id="target-order-index"', html_page.text) self.assertIn('id="target-capacity-songs"', html_page.text) self.assertIn("Save Target", html_page.text) self.assertIn('id="target-test-button"', html_page.text) self.assertIn("Test Connection", html_page.text) self.assertIn("Song Name", html_page.text) self.assertIn("External URL", html_page.text) self.assertNotIn("Secrets JSON", html_page.text) self.assertIn('id="sftp-host"', html_page.text) self.assertIn('id="sftp-remote-root"', html_page.text) self.assertIn('id="sftp-username"', html_page.text) self.assertIn('id="sftp-password"', html_page.text) self.assertIn('id="s3-bucket"', html_page.text) self.assertIn('id="s3-access-key-id"', html_page.text) self.assertIn('id="s3-secret-access-key"', html_page.text) def test_admin_can_test_unsaved_target_connection(self): with tempfile.TemporaryDirectory() as tmpdir: player_db_path = Path(tmpdir) / "player.db" catalog_db_path = Path(tmpdir) / "catalog_read.db" self._prepare_catalog_db(catalog_db_path) with patch.dict("os.environ", self._admin_env(player_db_path, catalog_db_path), clear=False): client = TestClient(create_app()) login = client.post( "/admin/session/login", data={"username": "admin", "password": "secret123"}, follow_redirects=False, ) self.assertIn(login.status_code, {200, 204, 303}) with patch.object( CacheService, "test_target_connection_payload", create=True, return_value={ "kind": "sftp", "ok": True, "secret_keys": ["host", "password", "username"], }, ) as mocked_test: response = client.post( "/admin/api/cache/targets/test", json={ "kind": "sftp", "secrets": { "host": "1.2.3.4", "port": 22, "username": "root", "password": "secret", }, }, ) self.assertEqual(200, response.status_code) self.assertEqual("sftp", response.json()["kind"]) self.assertTrue(response.json()["ok"]) mocked_test.assert_called_once_with( kind="sftp", secrets={ "host": "1.2.3.4", "port": 22, "username": "root", "password": "secret", }, ) if __name__ == "__main__": unittest.main()