"""Lyrics service (§6.7) + OpenSubsonic structured-lyrics serializer — DB-free. Service: get-or-fetch caching, not_found TTL, force refetch, graceful miss. Serializer: LRC → timed lines, plain fallback, empty payload when absent. """ import datetime as dt import uuid import pytest from app.api.rest.serializers import _parse_lrc, structured_lyrics from app.application.lyrics_service import LyricsService from app.domain.entities import Artist, Track from app.domain.entities.album import Album from app.domain.entities.lyrics import Lyrics, LyricsResult from app.domain.errors import NotFoundError def _now() -> dt.datetime: return dt.datetime.now(dt.UTC) def _lyrics(*, synced=None, plain=None, source=None, status="found", age_days=0) -> Lyrics: return Lyrics( track_id=uuid.uuid4(), synced=synced, plain=plain, source=source, status=status, fetched_at=_now() - dt.timedelta(days=age_days), ) def _track() -> Track: now = _now() return Track( id=uuid.uuid4(), title="Song", artist_id=uuid.uuid4(), album_id=None, storage_uri="tracks/x.mp3", file_format="mp3", file_size=1, source="upload", source_id="x", duration_seconds=200, genre=None, year=None, track_number=None, metadata_status="pending", metadata_error=None, enriched_at=None, availability="local", created_at=now, updated_at=now, ) class FakeLyricsRepo: def __init__(self, cached: Lyrics | None) -> None: self._cached = cached self.upserts: list[dict[str, object]] = [] async def get(self, track_id: uuid.UUID) -> Lyrics | None: return self._cached async def upsert(self, *, track_id, synced, plain, source, status) -> Lyrics: self.upserts.append({"status": status, "source": source}) return Lyrics( track_id=track_id, synced=synced, plain=plain, source=source, status=status, fetched_at=_now(), ) class FakeTrackRepo: def __init__(self, track: Track | None) -> None: self._track = track async def get_by_id(self, track_id: uuid.UUID) -> Track | None: return self._track class FakeArtistRepo: def __init__(self, artist: Artist | None = None) -> None: self._artist = artist async def get_by_id(self, artist_id: uuid.UUID) -> Artist | None: return self._artist class FakeAlbumRepo: async def get_by_id(self, album_id: uuid.UUID) -> Album | None: return None class FakeProvider: def __init__(self, result: LyricsResult | None) -> None: self._result = result self.calls = 0 async def fetch(self, *, artist, title, album, duration_seconds) -> LyricsResult | None: self.calls += 1 return self._result def _service(*, cached, track, provider) -> tuple[LyricsService, FakeLyricsRepo, FakeProvider]: repo = FakeLyricsRepo(cached) prov = FakeProvider(provider) svc = LyricsService( lyrics=repo, tracks=FakeTrackRepo(track), artists=FakeArtistRepo(), albums=FakeAlbumRepo(), provider=prov, ) return svc, repo, prov # -- service ------------------------------------------------------------------ async def test_fresh_found_is_served_from_cache_without_provider() -> None: cached = _lyrics(plain="hi", status="found") svc, _repo, prov = _service(cached=cached, track=_track(), provider=None) out = await svc.get_lyrics(cached.track_id) assert out is cached assert prov.calls == 0 async def test_miss_fetches_and_caches_found() -> None: track = _track() result = LyricsResult(synced="[00:01.00]hi", plain="hi", source="lrclib") svc, repo, prov = _service(cached=None, track=track, provider=result) out = await svc.get_lyrics(track.id) assert prov.calls == 1 assert out.status == "found" assert repo.upserts[-1]["status"] == "found" async def test_provider_miss_caches_not_found() -> None: track = _track() svc, repo, prov = _service(cached=None, track=track, provider=None) out = await svc.get_lyrics(track.id) assert prov.calls == 1 assert out.status == "not_found" assert repo.upserts[-1]["status"] == "not_found" async def test_stale_not_found_is_refetched() -> None: # A not_found older than the 7-day TTL is retried against the provider. stale = _lyrics(status="not_found", age_days=8) result = LyricsResult(synced=None, plain="found now", source="lrclib") svc, _repo, prov = _service(cached=stale, track=_track(), provider=result) out = await svc.get_lyrics(stale.track_id) assert prov.calls == 1 assert out.status == "found" async def test_fresh_not_found_is_not_refetched() -> None: fresh = _lyrics(status="not_found", age_days=1) svc, _repo, prov = _service(cached=fresh, track=_track(), provider=None) out = await svc.get_lyrics(fresh.track_id) assert out is fresh assert prov.calls == 0 async def test_force_bypasses_a_fresh_cache() -> None: cached = _lyrics(plain="old", status="found") result = LyricsResult(synced=None, plain="new", source="lrclib") svc, _repo, prov = _service(cached=cached, track=_track(), provider=result) out = await svc.get_lyrics(cached.track_id, force=True) assert prov.calls == 1 assert out.plain == "new" async def test_unknown_track_raises() -> None: svc, _repo, _prov = _service(cached=None, track=None, provider=None) with pytest.raises(NotFoundError): await svc.get_lyrics(uuid.uuid4()) # -- serializer --------------------------------------------------------------- def test_parse_lrc_times_and_order() -> None: lines = _parse_lrc("[00:12.34]second\n[00:01.00]first\n[bad]meta\n\n") # Metadata/blank lines dropped; output ordered by time; centiseconds → ms. assert lines == [ {"start": 1000, "value": "first"}, {"start": 12340, "value": "second"}, ] def test_parse_lrc_repeated_stamps_on_one_line() -> None: lines = _parse_lrc("[00:01.00][00:05.00]chorus") assert lines == [ {"start": 1000, "value": "chorus"}, {"start": 5000, "value": "chorus"}, ] def test_structured_lyrics_prefers_synced() -> None: out = structured_lyrics( _lyrics(synced="[00:01.00]hi", plain="hi", status="found"), display_artist="A", display_title="T", ) entry = out["lyricsList"]["structuredLyrics"][0] assert entry["synced"] is True assert entry["displayArtist"] == "A" assert entry["line"] == [{"start": 1000, "value": "hi"}] def test_structured_lyrics_falls_back_to_plain() -> None: out = structured_lyrics( _lyrics(synced=None, plain="line one\nline two", status="found"), display_artist="A", display_title="T", ) entry = out["lyricsList"]["structuredLyrics"][0] assert entry["synced"] is False assert entry["line"] == [{"value": "line one"}, {"value": "line two"}] def test_structured_lyrics_empty_when_absent() -> None: out = structured_lyrics( _lyrics(synced=None, plain=None, status="not_found"), display_artist="A", display_title="T", ) assert out == {"lyricsList": {}} def test_structured_lyrics_unparseable_synced_falls_back_to_plain() -> None: # synced present but no valid timecodes → treat as plain, synced=False. out = structured_lyrics( _lyrics(synced="no timecodes here", plain="plain text", status="found"), display_artist="A", display_title="T", ) entry = out["lyricsList"]["structuredLyrics"][0] assert entry["synced"] is False assert entry["line"] == [{"value": "plain text"}]