Senko-san
78007461e1
Pluggable fetch source: ytmusicapi search + yt-dlp download (cookies-file guard), DownloadJob entity/repo + DownloadService, download_task worker with exponential-backoff retries, and wired /search, /sources/{source}/search, and /downloads endpoints. Adds youtube_enabled/cookies config, yt-dlp+ytmusicapi deps, and the download_jobs.track_id migration. Snapshot also bundles in-progress storage/tracks/acoustid edits.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
"""Unit tests for YouTubeMusicSource + registry (no network, injected libs)."""
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from app.core.config import Settings
|
|
from app.domain.sources import KIND_FETCH
|
|
from app.infrastructure.sources.registry import build_source_registry
|
|
from app.infrastructure.sources.youtube import YouTubeMusicSource
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
def _song_row(**overrides: Any) -> dict[str, Any]:
|
|
row: dict[str, Any] = {
|
|
"videoId": "abc123",
|
|
"title": "Bohemian Rhapsody",
|
|
"artists": [{"name": "Queen", "id": "a1"}],
|
|
"album": {"name": "A Night at the Opera", "id": "al1"},
|
|
"duration_seconds": 354,
|
|
"thumbnails": [
|
|
{"url": "http://img/small.jpg", "width": 60, "height": 60},
|
|
{"url": "http://img/large.jpg", "width": 240, "height": 240},
|
|
],
|
|
}
|
|
row.update(overrides)
|
|
return row
|
|
|
|
|
|
def _settings(**overrides: object) -> Settings:
|
|
return Settings(**overrides) # type: ignore[arg-type]
|
|
|
|
|
|
async def test_search_maps_ytmusic_rows() -> None:
|
|
source = YouTubeMusicSource(search_fn=lambda q, limit: [_song_row()])
|
|
[result] = await source.search("queen", limit=10)
|
|
|
|
assert result.source == "youtube"
|
|
assert result.source_id == "abc123"
|
|
assert result.title == "Bohemian Rhapsody"
|
|
assert result.artist == "Queen"
|
|
assert result.album == "A Night at the Opera"
|
|
assert result.duration_seconds == 354
|
|
assert result.thumbnail_url == "http://img/large.jpg" # last (largest)
|
|
|
|
|
|
async def test_search_joins_multiple_artists_and_tolerates_missing_fields() -> None:
|
|
row = _song_row(
|
|
artists=[{"name": "Queen"}, {"name": "David Bowie"}],
|
|
album=None,
|
|
thumbnails=[],
|
|
duration_seconds=None,
|
|
)
|
|
source = YouTubeMusicSource(search_fn=lambda q, limit: [row])
|
|
[result] = await source.search("under pressure", limit=10)
|
|
|
|
assert result.artist == "Queen, David Bowie"
|
|
assert result.album is None
|
|
assert result.thumbnail_url is None
|
|
assert result.duration_seconds is None
|
|
|
|
|
|
async def test_search_drops_rows_without_video_id() -> None:
|
|
rows = [_song_row(), _song_row(videoId=None), _song_row(videoId="xyz")]
|
|
source = YouTubeMusicSource(search_fn=lambda q, limit: rows)
|
|
results = await source.search("q", limit=10)
|
|
assert [r.source_id for r in results] == ["abc123", "xyz"]
|
|
|
|
|
|
async def test_search_empty_query_short_circuits() -> None:
|
|
called = False
|
|
|
|
def _search(q: str, limit: int) -> list[dict[str, Any]]:
|
|
nonlocal called
|
|
called = True
|
|
return []
|
|
|
|
source = YouTubeMusicSource(search_fn=_search)
|
|
assert await source.search(" ", limit=10) == []
|
|
assert called is False
|
|
|
|
|
|
async def test_search_degrades_to_empty_on_error() -> None:
|
|
def _boom(q: str, limit: int) -> list[dict[str, Any]]:
|
|
raise RuntimeError("service down")
|
|
|
|
source = YouTubeMusicSource(search_fn=_boom)
|
|
assert await source.search("q", limit=10) == []
|
|
|
|
|
|
async def test_fetch_maps_download_result(tmp_path: Path) -> None:
|
|
audio = tmp_path / "abc123.m4a"
|
|
audio.write_bytes(b"opus-bytes" * 10)
|
|
|
|
def _download(video_id: str, tmp_dir: Path, hook: Any, cookies: Path | None) -> dict[str, Any]:
|
|
return {
|
|
"filepath": audio,
|
|
"file_format": "m4a",
|
|
"bitrate": 160,
|
|
"title": "Bohemian Rhapsody",
|
|
}
|
|
|
|
source = YouTubeMusicSource(download_fn=_download)
|
|
result = await source.fetch("abc123")
|
|
|
|
assert result.source_id == "abc123"
|
|
assert result.path == audio
|
|
assert result.file_format == "m4a"
|
|
assert result.file_size == len(b"opus-bytes" * 10)
|
|
assert result.bitrate == 160
|
|
assert result.suggested_title == "Bohemian Rhapsody"
|
|
|
|
|
|
async def test_info_and_availability_with_injected_fn() -> None:
|
|
source = YouTubeMusicSource(search_fn=lambda q, limit: [])
|
|
info = source.info()
|
|
assert info.name == "youtube"
|
|
assert info.kind == KIND_FETCH
|
|
assert info.available is True # injected fn → treated as available
|
|
|
|
|
|
async def test_registry_registers_youtube_when_enabled() -> None:
|
|
registry = build_source_registry(_settings(youtube_enabled=True))
|
|
names = {info.name for info in registry.infos()}
|
|
assert "youtube" in names
|
|
# youtube is searchable + fetchable, not indexable
|
|
assert registry.searchable("youtube").name == "youtube"
|
|
assert registry.fetchable("youtube").name == "youtube"
|
|
|
|
|
|
async def test_registry_omits_youtube_when_disabled() -> None:
|
|
registry = build_source_registry(_settings(youtube_enabled=False))
|
|
names = {info.name for info in registry.infos()}
|
|
assert "youtube" not in names
|