478 lines
17 KiB
Python
478 lines
17 KiB
Python
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import closing
|
|
from pathlib import Path
|
|
from unittest.mock import ANY, patch
|
|
|
|
from click.testing import CliRunner
|
|
|
|
|
|
class CatalogCliTests(unittest.TestCase):
|
|
def test_init_db_command_creates_sqlite_file(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
library_root = Path(tmpdir) / "library"
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
["init-db", "--db", str(db_path), "--library-root", str(library_root)],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
self.assertTrue(db_path.exists())
|
|
with closing(sqlite3.connect(db_path)) as conn:
|
|
table_names = {
|
|
row[0]
|
|
for row in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
|
).fetchall()
|
|
}
|
|
self.assertIn("songs", table_names)
|
|
|
|
def test_init_db_command_creates_resolver_stats_side_db(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
from musicdl.catalogsync.resolver_stats import default_resolver_stats_db_path
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
library_root = Path(tmpdir) / "library"
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
["init-db", "--db", str(db_path), "--library-root", str(library_root)],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
resolver_stats_db_path = default_resolver_stats_db_path(db_path)
|
|
self.assertTrue(resolver_stats_db_path.exists())
|
|
|
|
def test_run_command_wires_collect_sync_and_download_steps(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"run",
|
|
"--db",
|
|
str(db_path),
|
|
"--sources",
|
|
"netease,qq",
|
|
"--download-sources",
|
|
"qq,kuwo,migu",
|
|
"--library-root",
|
|
str(Path(tmpdir) / "library"),
|
|
"--workers",
|
|
"3",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.collect_playlists.assert_called_once()
|
|
app.sync_playlist_catalog.assert_called_once()
|
|
app.download_pending.assert_called_once_with(
|
|
["netease", "qq"],
|
|
limit=None,
|
|
workers=3,
|
|
download_sources=["qq", "kuwo", "migu"],
|
|
lyrics_enabled=True,
|
|
overwrite_lyrics=False,
|
|
)
|
|
|
|
def test_run_command_uses_playlist_file_branch_without_collect(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
playlist_file = Path(tmpdir) / "playlists.txt"
|
|
playlist_file.write_text(
|
|
"https://music.163.com/#/playlist?id=17745989905\n",
|
|
encoding="utf-8",
|
|
)
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"run",
|
|
"--db",
|
|
str(db_path),
|
|
"--library-root",
|
|
str(Path(tmpdir) / "library"),
|
|
"--playlist-file",
|
|
str(playlist_file),
|
|
"--download-sources",
|
|
"qq,kuwo",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.collect_playlists.assert_not_called()
|
|
app.run_playlist_file.assert_called_once_with(
|
|
playlist_file=str(playlist_file),
|
|
limit=None,
|
|
workers=10,
|
|
download_sources=["qq", "kuwo"],
|
|
lyrics_enabled=True,
|
|
overwrite_lyrics=False,
|
|
)
|
|
|
|
def test_download_command_defaults_workers_to_ten(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"download",
|
|
"--db",
|
|
str(db_path),
|
|
"--sources",
|
|
"netease,qq",
|
|
"--download-sources",
|
|
"qq,kuwo,migu",
|
|
"--library-root",
|
|
str(Path(tmpdir) / "library"),
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.download_pending.assert_called_once_with(
|
|
["netease", "qq"],
|
|
limit=None,
|
|
workers=10,
|
|
download_sources=["qq", "kuwo", "migu"],
|
|
lyrics_enabled=True,
|
|
overwrite_lyrics=False,
|
|
)
|
|
|
|
def test_download_command_reads_workers_from_download_workers_env(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"download",
|
|
"--db",
|
|
str(db_path),
|
|
"--sources",
|
|
"netease,qq",
|
|
"--download-sources",
|
|
"qq,kuwo,migu",
|
|
"--library-root",
|
|
str(Path(tmpdir) / "library"),
|
|
],
|
|
env={"DOWNLOAD_WORKERS": "8"},
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.download_pending.assert_called_once_with(
|
|
["netease", "qq"],
|
|
limit=None,
|
|
workers=8,
|
|
download_sources=["qq", "kuwo", "migu"],
|
|
lyrics_enabled=True,
|
|
overwrite_lyrics=False,
|
|
)
|
|
|
|
def test_download_command_forwards_workers(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"download",
|
|
"--db",
|
|
str(db_path),
|
|
"--sources",
|
|
"netease,qq",
|
|
"--download-sources",
|
|
"qq,kuwo,migu",
|
|
"--library-root",
|
|
str(Path(tmpdir) / "library"),
|
|
"--workers",
|
|
"5",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.download_pending.assert_called_once_with(
|
|
["netease", "qq"],
|
|
limit=None,
|
|
workers=5,
|
|
download_sources=["qq", "kuwo", "migu"],
|
|
lyrics_enabled=True,
|
|
overwrite_lyrics=False,
|
|
)
|
|
|
|
def test_download_command_forwards_lyrics_flags(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"download",
|
|
"--db",
|
|
str(db_path),
|
|
"--sources",
|
|
"netease",
|
|
"--download-sources",
|
|
"qq",
|
|
"--library-root",
|
|
str(Path(tmpdir) / "library"),
|
|
"--no-lyrics",
|
|
"--overwrite-lyrics",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.download_pending.assert_called_once_with(
|
|
["netease"],
|
|
limit=None,
|
|
workers=10,
|
|
download_sources=["qq"],
|
|
lyrics_enabled=False,
|
|
overwrite_lyrics=True,
|
|
)
|
|
|
|
def test_lyrics_command_wires_application_method_and_filters(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
def sync_local_lyrics_side_effect(*args, **kwargs):
|
|
progress_callback = kwargs["progress_callback"]
|
|
progress_callback(
|
|
total=10,
|
|
processed=0,
|
|
saved=0,
|
|
skipped=0,
|
|
failed=0,
|
|
progress_percent=0,
|
|
)
|
|
progress_callback(
|
|
total=10,
|
|
processed=10,
|
|
saved=7,
|
|
skipped=2,
|
|
failed=1,
|
|
progress_percent=100,
|
|
)
|
|
return {"total": 10, "processed": 10, "saved": 7, "skipped": 2, "failed": 1}
|
|
|
|
app.sync_local_lyrics.side_effect = sync_local_lyrics_side_effect
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"lyrics",
|
|
"--db",
|
|
str(db_path),
|
|
"--sources",
|
|
"netease,qq",
|
|
"--playlist-ids",
|
|
"12,15",
|
|
"--limit",
|
|
"200",
|
|
"--workers",
|
|
"8",
|
|
"--overwrite-lyrics",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.sync_local_lyrics.assert_called_once_with(
|
|
sources=["netease", "qq"],
|
|
playlist_ids=[12, 15],
|
|
limit=200,
|
|
workers=8,
|
|
progress_callback=ANY,
|
|
overwrite_lyrics=True,
|
|
)
|
|
self.assertIn("Lyrics progress: 0/10 (0%)", result.output)
|
|
self.assertIn("Lyrics progress: 10/10 (100%)", result.output)
|
|
|
|
def test_register_object_backend_command_wires_application_method(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"register-object-backend",
|
|
"--db",
|
|
str(db_path),
|
|
"--backend",
|
|
"main-s3",
|
|
"--bucket",
|
|
"music-bucket",
|
|
"--endpoint",
|
|
"https://s3.example.com",
|
|
"--region",
|
|
"auto",
|
|
"--base-prefix",
|
|
"music",
|
|
"--credential-env-prefix",
|
|
"CATALOGSYNC_MAIN_S3",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.register_object_backend.assert_called_once_with(
|
|
backend_name="main-s3",
|
|
container_name="music-bucket",
|
|
endpoint="https://s3.example.com",
|
|
region="auto",
|
|
base_prefix="music",
|
|
credential_env_prefix="CATALOGSYNC_MAIN_S3",
|
|
addressing_style=None,
|
|
public_base_url=None,
|
|
)
|
|
|
|
def test_upload_command_wires_application_method_and_filters(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
with patch("musicdl.catalogsync.cli.CatalogSyncApplication") as app_cls:
|
|
app = app_cls.return_value
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"upload",
|
|
"--db",
|
|
str(db_path),
|
|
"--backend",
|
|
"main-s3",
|
|
"--sources",
|
|
"netease,qq",
|
|
"--playlist-ids",
|
|
"12,15",
|
|
"--limit",
|
|
"200",
|
|
"--workers",
|
|
"4",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
app.upload_files.assert_called_once_with(
|
|
backend_name="main-s3",
|
|
sources=["netease", "qq"],
|
|
playlist_ids=[12, 15],
|
|
limit=200,
|
|
workers=4,
|
|
)
|
|
|
|
def test_serve_command_wires_ops_web_app_and_uvicorn(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
env_file = Path(tmpdir) / "catalogsync.env"
|
|
fake_app = object()
|
|
with patch(
|
|
"musicdl.catalogsync.cli.create_ops_web_app",
|
|
return_value=fake_app,
|
|
) as create_app_mock, patch("musicdl.catalogsync.cli.uvicorn.run") as uvicorn_run_mock:
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"serve",
|
|
"--db",
|
|
str(db_path),
|
|
"--env-file",
|
|
str(env_file),
|
|
"--host",
|
|
"0.0.0.0",
|
|
"--port",
|
|
"19090",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code, msg=result.output)
|
|
create_app_mock.assert_called_once_with(
|
|
db_path=str(db_path),
|
|
env_path=str(env_file),
|
|
)
|
|
uvicorn_run_mock.assert_called_once_with(
|
|
fake_app,
|
|
host="0.0.0.0",
|
|
port=19090,
|
|
)
|
|
|
|
def test_serve_command_rejects_out_of_range_port(self):
|
|
from musicdl.catalogsync.cli import cli
|
|
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
db_path = Path(tmpdir) / "catalogsync.db"
|
|
env_file = Path(tmpdir) / "catalogsync.env"
|
|
with patch("musicdl.catalogsync.cli.create_ops_web_app") as create_app_mock:
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"serve",
|
|
"--db",
|
|
str(db_path),
|
|
"--env-file",
|
|
str(env_file),
|
|
"--port",
|
|
"70000",
|
|
],
|
|
)
|
|
|
|
self.assertNotEqual(0, result.exit_code)
|
|
self.assertIn("is not in the range", result.output)
|
|
create_app_mock.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|