test: cover lyrics, transcode, and recommendation
DB-free unit tests for the three previously-untested features: - lyrics: get-or-fetch caching, not_found TTL, force refetch, graceful miss; plus the LRC -> structured-lyrics serializer (timing, plain fallback, empty) - transcode: path helpers, segment-name traversal guard, cache hit/miss, the unknown/not-downloaded 404 paths - reco: ML path (order-preserving batched hydration) + metadata fallback for similar/radio, exclude handling, reason codes Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Цвылев Александр Вадимович
@@ -0,0 +1,247 @@
|
||||
"""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"}]
|
||||
@@ -0,0 +1,248 @@
|
||||
"""RecommendationService (§6.5) — DB-free, in-memory fakes.
|
||||
|
||||
Covers the two paths that matter: the ML recommender when available (reason
|
||||
``ml``, order-preserving batched hydration) and the metadata fallback when it
|
||||
declines (similar/radio still work — the graceful-degradation invariant).
|
||||
"""
|
||||
|
||||
import datetime as dt
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from app.application.recommendation_service import (
|
||||
REASON_FROM_LIKES,
|
||||
REASON_ML,
|
||||
REASON_SIMILAR,
|
||||
RecommendationService,
|
||||
)
|
||||
from app.domain.entities import Artist, Track
|
||||
from app.domain.errors import NotFoundError
|
||||
from app.infrastructure.ml.recommender import NullRecommender
|
||||
|
||||
|
||||
def _now() -> dt.datetime:
|
||||
return dt.datetime.now(dt.UTC)
|
||||
|
||||
|
||||
def _track(*, genre: str | None = "rock", artist_id: uuid.UUID | None = None) -> Track:
|
||||
now = _now()
|
||||
return Track(
|
||||
id=uuid.uuid4(),
|
||||
title="T",
|
||||
artist_id=artist_id or uuid.uuid4(),
|
||||
album_id=None,
|
||||
storage_uri="tracks/x.mp3",
|
||||
file_format="mp3",
|
||||
file_size=1,
|
||||
source="upload",
|
||||
source_id="x",
|
||||
duration_seconds=None,
|
||||
genre=genre,
|
||||
year=None,
|
||||
track_number=None,
|
||||
metadata_status="pending",
|
||||
metadata_error=None,
|
||||
enriched_at=None,
|
||||
availability="local",
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
|
||||
|
||||
def _artist() -> Artist:
|
||||
now = _now()
|
||||
return Artist(
|
||||
id=uuid.uuid4(), name="A", source=None, source_id=None, created_at=now, updated_at=now
|
||||
)
|
||||
|
||||
|
||||
class FakeTrackRepo:
|
||||
def __init__(self, tracks: list[Track]) -> None:
|
||||
self._by_id = {t.id: t for t in tracks}
|
||||
self.similar: list[Track] = []
|
||||
self.sample: list[Track] = []
|
||||
self.similar_calls: list[dict[str, object]] = []
|
||||
|
||||
async def get_by_id(self, track_id: uuid.UUID) -> Track | None:
|
||||
return self._by_id.get(track_id)
|
||||
|
||||
async def get_many(self, ids: list[uuid.UUID]) -> list[Track]:
|
||||
# Deliberately unordered (mirrors a real ``WHERE id IN`` query) so the
|
||||
# service is responsible for restoring request order.
|
||||
return [self._by_id[i] for i in reversed(ids) if i in self._by_id]
|
||||
|
||||
async def list_similar(self, *, genre, artist_id, exclude_ids, limit) -> list[Track]:
|
||||
self.similar_calls.append({"exclude_ids": list(exclude_ids), "limit": limit})
|
||||
return [t for t in self.similar if t.id not in exclude_ids][:limit]
|
||||
|
||||
async def sample_playable(self, *, exclude_ids, limit) -> list[Track]:
|
||||
return [t for t in self.sample if t.id not in exclude_ids][:limit]
|
||||
|
||||
|
||||
class FakeArtistRepo:
|
||||
def __init__(self, artists: list[Artist]) -> None:
|
||||
self._by_id = {a.id: a for a in artists}
|
||||
self.similar: list[Artist] = []
|
||||
|
||||
async def get_by_id(self, artist_id: uuid.UUID) -> Artist | None:
|
||||
return self._by_id.get(artist_id)
|
||||
|
||||
async def get_many(self, ids: list[uuid.UUID]) -> list[Artist]:
|
||||
return [self._by_id[i] for i in reversed(ids) if i in self._by_id]
|
||||
|
||||
async def list_similar(self, *, artist_id, limit) -> list[Artist]:
|
||||
return self.similar[:limit]
|
||||
|
||||
|
||||
class FakeLikeRepo:
|
||||
def __init__(self, liked: list[Track]) -> None:
|
||||
self.liked = liked
|
||||
|
||||
async def list_liked_tracks(self, *, user_id, limit, offset) -> list[Track]:
|
||||
return self.liked[offset : offset + limit]
|
||||
|
||||
|
||||
class StubRecommender:
|
||||
"""An 'available' ML recommender returning fixed ids for the ML path."""
|
||||
|
||||
def __init__(self, ids: list[uuid.UUID] | None) -> None:
|
||||
self._ids = ids
|
||||
|
||||
def is_available(self) -> bool:
|
||||
return True
|
||||
|
||||
async def similar_track_ids(self, track_id, *, limit, exclude_ids):
|
||||
return self._ids
|
||||
|
||||
async def similar_artist_ids(self, artist_id, *, limit):
|
||||
return self._ids
|
||||
|
||||
async def radio_track_ids(self, *, seed_track_id, exploration, limit, exclude_ids):
|
||||
return self._ids
|
||||
|
||||
|
||||
def _service(tracks, artists, likes, recommender) -> RecommendationService:
|
||||
return RecommendationService(
|
||||
recommender=recommender, tracks=tracks, artists=artists, likes=likes
|
||||
)
|
||||
|
||||
|
||||
# -- similar ------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_similar_tracks_unknown_seed_raises() -> None:
|
||||
svc = _service(FakeTrackRepo([]), FakeArtistRepo([]), FakeLikeRepo([]), NullRecommender())
|
||||
with pytest.raises(NotFoundError):
|
||||
await svc.similar_tracks(uuid.uuid4(), limit=5)
|
||||
|
||||
|
||||
async def test_similar_tracks_falls_back_to_metadata() -> None:
|
||||
seed = _track()
|
||||
neighbours = [_track(), _track()]
|
||||
repo = FakeTrackRepo([seed, *neighbours])
|
||||
repo.similar = neighbours
|
||||
svc = _service(repo, FakeArtistRepo([]), FakeLikeRepo([]), NullRecommender())
|
||||
|
||||
reason, found = await svc.similar_tracks(seed.id, limit=5)
|
||||
assert reason == REASON_SIMILAR
|
||||
assert [t.id for t in found] == [n.id for n in neighbours]
|
||||
# The seed itself is always excluded from its own neighbours.
|
||||
assert repo.similar_calls[-1]["exclude_ids"] == [seed.id]
|
||||
|
||||
|
||||
async def test_similar_tracks_uses_ml_and_preserves_order() -> None:
|
||||
seed = _track()
|
||||
a, b, c = _track(), _track(), _track()
|
||||
repo = FakeTrackRepo([seed, a, b, c])
|
||||
# ML returns a specific order; hydration must preserve it despite get_many
|
||||
# returning rows unordered, and skip ids that no longer exist.
|
||||
ml_ids = [c.id, uuid.uuid4(), a.id, b.id]
|
||||
svc = _service(repo, FakeArtistRepo([]), FakeLikeRepo([]), StubRecommender(ml_ids))
|
||||
|
||||
reason, found = await svc.similar_tracks(seed.id, limit=10)
|
||||
assert reason == REASON_ML
|
||||
assert [t.id for t in found] == [c.id, a.id, b.id]
|
||||
|
||||
|
||||
async def test_similar_artists_falls_back_to_metadata() -> None:
|
||||
seed = _artist()
|
||||
neighbours = [_artist(), _artist()]
|
||||
repo = FakeArtistRepo([seed])
|
||||
repo.similar = neighbours
|
||||
svc = _service(FakeTrackRepo([]), repo, FakeLikeRepo([]), NullRecommender())
|
||||
|
||||
reason, found = await svc.similar_artists(seed.id, limit=5)
|
||||
assert reason == REASON_SIMILAR
|
||||
assert [a.id for a in found] == [n.id for n in neighbours]
|
||||
|
||||
|
||||
# -- radio --------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_radio_from_likes_seeds_and_fills(monkeypatch) -> None:
|
||||
liked = _track()
|
||||
similar = [_track(), _track()]
|
||||
explore = [_track(), _track(), _track()]
|
||||
repo = FakeTrackRepo([liked, *similar, *explore])
|
||||
repo.similar = similar
|
||||
repo.sample = explore
|
||||
svc = _service(repo, FakeArtistRepo([]), FakeLikeRepo([liked]), NullRecommender())
|
||||
# Deterministic seed choice + no shuffling for a stable assertion.
|
||||
monkeypatch.setattr("app.application.recommendation_service.random.choice", lambda s: s[0])
|
||||
monkeypatch.setattr("app.application.recommendation_service.random.shuffle", lambda s: None)
|
||||
|
||||
reason, picks = await svc.radio(
|
||||
user_id=uuid.uuid4(),
|
||||
seed_track_id=None,
|
||||
from_likes=True,
|
||||
exploration=0.5,
|
||||
limit=4,
|
||||
exclude_ids=[],
|
||||
)
|
||||
assert reason == "metadata"
|
||||
ids = [p.track.id for p in picks]
|
||||
# 4 picks, all distinct, no seed repeated, mix of similar + discover.
|
||||
assert len(ids) == 4
|
||||
assert len(set(ids)) == 4
|
||||
reasons = {p.reason for p in picks}
|
||||
assert REASON_FROM_LIKES in reasons # similar picks carry the seed's reason
|
||||
|
||||
|
||||
async def test_radio_respects_exclude_ids(monkeypatch) -> None:
|
||||
similar = [_track(), _track()]
|
||||
explore = [_track(), _track()]
|
||||
repo = FakeTrackRepo([*similar, *explore])
|
||||
repo.similar = similar
|
||||
repo.sample = explore
|
||||
svc = _service(repo, FakeArtistRepo([]), FakeLikeRepo([]), NullRecommender())
|
||||
monkeypatch.setattr("app.application.recommendation_service.random.shuffle", lambda s: None)
|
||||
|
||||
excluded = similar[0].id
|
||||
_, picks = await svc.radio(
|
||||
user_id=uuid.uuid4(),
|
||||
seed_track_id=None,
|
||||
from_likes=False,
|
||||
exploration=1.0,
|
||||
limit=4,
|
||||
exclude_ids=[excluded],
|
||||
)
|
||||
assert excluded not in {p.track.id for p in picks}
|
||||
|
||||
|
||||
async def test_radio_uses_ml_when_available(monkeypatch) -> None:
|
||||
a, b = _track(), _track()
|
||||
repo = FakeTrackRepo([a, b])
|
||||
svc = _service(repo, FakeArtistRepo([]), FakeLikeRepo([]), StubRecommender([b.id, a.id]))
|
||||
monkeypatch.setattr("app.application.recommendation_service.random.shuffle", lambda s: None)
|
||||
|
||||
reason, picks = await svc.radio(
|
||||
user_id=uuid.uuid4(),
|
||||
seed_track_id=a.id,
|
||||
from_likes=False,
|
||||
exploration=0.3,
|
||||
limit=5,
|
||||
exclude_ids=[],
|
||||
)
|
||||
assert reason == REASON_ML
|
||||
assert [p.track.id for p in picks] == [b.id, a.id]
|
||||
assert all(p.reason == REASON_ML for p in picks)
|
||||
@@ -0,0 +1,157 @@
|
||||
"""Transcode cache-path helpers + request-side service (§6.6) — DB-free.
|
||||
|
||||
Covers the pure path math, the segment-name traversal guard, cache
|
||||
hit/miss lookups, and the "not yet downloaded" / unknown-track 404 paths.
|
||||
"""
|
||||
|
||||
import datetime as dt
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from app.application.transcode_service import (
|
||||
QUALITY_BITRATE,
|
||||
TranscodeService,
|
||||
bitrate_for_quality,
|
||||
hls_playlist_path,
|
||||
hls_segment_path,
|
||||
opus_path,
|
||||
remove_track_cache,
|
||||
track_cache_dir,
|
||||
)
|
||||
from app.domain.entities import Track
|
||||
from app.domain.errors import NotFoundError
|
||||
|
||||
|
||||
def _track(*, storage_uri: str | None = "tracks/aa/song.mp3") -> Track:
|
||||
now = dt.datetime.now(dt.UTC)
|
||||
return Track(
|
||||
id=uuid.uuid4(),
|
||||
title="Song",
|
||||
artist_id=uuid.uuid4(),
|
||||
album_id=None,
|
||||
storage_uri=storage_uri,
|
||||
file_format="mp3",
|
||||
file_size=1,
|
||||
source="upload",
|
||||
source_id="x",
|
||||
duration_seconds=None,
|
||||
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 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
|
||||
|
||||
|
||||
# -- pure helpers -------------------------------------------------------------
|
||||
|
||||
|
||||
def test_bitrate_for_quality_known_and_unknown() -> None:
|
||||
assert bitrate_for_quality("high") == QUALITY_BITRATE["high"]
|
||||
assert bitrate_for_quality("medium") == 96
|
||||
assert bitrate_for_quality("low") == 64
|
||||
# "original" and anything unrecognised mean "serve the master, no transcode".
|
||||
assert bitrate_for_quality("original") is None
|
||||
assert bitrate_for_quality("nonsense") is None
|
||||
|
||||
|
||||
def test_path_helpers_are_under_the_track_dir(tmp_path: Path) -> None:
|
||||
tid = uuid.uuid4()
|
||||
cache = track_cache_dir(tmp_path, tid)
|
||||
assert cache == tmp_path / str(tid)
|
||||
assert opus_path(tmp_path, tid, 128) == cache / "opus_128.opus"
|
||||
assert hls_playlist_path(tmp_path, tid) == cache / "hls" / "playlist.m3u8"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"name",
|
||||
["seg_000.ts", "seg_1234.ts"],
|
||||
)
|
||||
def test_hls_segment_path_accepts_valid_names(tmp_path: Path, name: str) -> None:
|
||||
tid = uuid.uuid4()
|
||||
resolved = hls_segment_path(tmp_path, tid, name)
|
||||
assert resolved == track_cache_dir(tmp_path, tid) / "hls" / name
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"name",
|
||||
["../../etc/passwd", "seg_.ts", "seg_00.tsx", "playlist.m3u8", "seg_00.ts", "..", ""],
|
||||
)
|
||||
def test_hls_segment_path_rejects_traversal_and_junk(tmp_path: Path, name: str) -> None:
|
||||
assert hls_segment_path(tmp_path, uuid.uuid4(), name) is None
|
||||
|
||||
|
||||
def test_remove_track_cache_is_best_effort(tmp_path: Path) -> None:
|
||||
tid = uuid.uuid4()
|
||||
d = track_cache_dir(tmp_path, tid)
|
||||
(d / "hls").mkdir(parents=True)
|
||||
(d / "opus_128.opus").write_bytes(b"x")
|
||||
remove_track_cache(tmp_path, tid)
|
||||
assert not d.exists()
|
||||
# Idempotent: deleting an already-absent cache doesn't raise.
|
||||
remove_track_cache(tmp_path, tid)
|
||||
|
||||
|
||||
# -- request-side service -----------------------------------------------------
|
||||
|
||||
|
||||
async def test_resolve_quality_file_hit_and_miss(tmp_path: Path) -> None:
|
||||
track = _track()
|
||||
svc = TranscodeService(tracks=FakeTrackRepo(track), cache_root=tmp_path)
|
||||
|
||||
# Miss: nothing on disk yet.
|
||||
assert await svc.resolve_quality_file(track.id, "high") is None
|
||||
# "original" never has a rendition.
|
||||
assert await svc.resolve_quality_file(track.id, "original") is None
|
||||
|
||||
# Hit: once the rendition exists it's returned.
|
||||
path = opus_path(tmp_path, track.id, 128)
|
||||
path.parent.mkdir(parents=True)
|
||||
path.write_bytes(b"opus")
|
||||
assert await svc.resolve_quality_file(track.id, "high") == path
|
||||
|
||||
|
||||
async def test_hls_playlist_requires_downloaded_track(tmp_path: Path) -> None:
|
||||
tid = uuid.uuid4()
|
||||
# Unknown track → 404.
|
||||
with pytest.raises(NotFoundError):
|
||||
await TranscodeService(tracks=FakeTrackRepo(None), cache_root=tmp_path).hls_playlist(tid)
|
||||
|
||||
# Known but not downloaded (no storage_uri) → 404.
|
||||
pending = _track(storage_uri=None)
|
||||
with pytest.raises(NotFoundError):
|
||||
await TranscodeService(tracks=FakeTrackRepo(pending), cache_root=tmp_path).hls_playlist(
|
||||
pending.id
|
||||
)
|
||||
|
||||
|
||||
async def test_hls_playlist_miss_then_hit(tmp_path: Path) -> None:
|
||||
track = _track()
|
||||
svc = TranscodeService(tracks=FakeTrackRepo(track), cache_root=tmp_path)
|
||||
assert await svc.hls_playlist(track.id) is None
|
||||
|
||||
playlist = hls_playlist_path(tmp_path, track.id)
|
||||
playlist.parent.mkdir(parents=True)
|
||||
playlist.write_text("#EXTM3U")
|
||||
assert await svc.hls_playlist(track.id) == playlist
|
||||
|
||||
|
||||
def test_hls_segment_returns_none_for_missing_file(tmp_path: Path) -> None:
|
||||
svc = TranscodeService(tracks=FakeTrackRepo(_track()), cache_root=tmp_path)
|
||||
# Valid name but no file on disk.
|
||||
assert svc.hls_segment(uuid.uuid4(), "seg_000.ts") is None
|
||||
# Junk name is refused outright.
|
||||
assert svc.hls_segment(uuid.uuid4(), "../secret") is None
|
||||
Reference in New Issue
Block a user