mcma-backend/tests/test_transcode_service.py

"""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