mcma-backend/tests/test_cover_api.py

"""Integration tests for the native cover-art endpoints.

Seeds an album with a stored cover, then exercises the ``/api/v1`` album/track
cover endpoints (token auth, 404 when absent). Requires a reachable Postgres;
skips otherwise.
"""

import asyncio
import os
import uuid
from collections.abc import AsyncIterator
from pathlib import Path

import pytest
from app.core.config import get_settings
from app.infrastructure.db import Base, dispose_engine, get_engine, session_scope
from app.infrastructure.db.repositories import (
    SqlAlchemyAlbumRepository,
    SqlAlchemyArtistRepository,
    SqlAlchemyRefreshTokenRepository,
    SqlAlchemyTrackRepository,
    SqlAlchemyUserRepository,
)
from app.infrastructure.storage.provider import get_file_storage
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient

pytestmark = pytest.mark.asyncio

# A minimal valid 1x1 PNG.
_PNG_BYTES = bytes.fromhex(
    "89504e470d0a1a0a0000000d4948445200000001000000010802000000907753"
    "de0000000c4944415408d763f8cfc0f01f0005000155a2b4f60000000049454e44ae426082"
)

_db_reachable_cache: bool | None = None


async def _db_reachable() -> bool:
    global _db_reachable_cache
    if _db_reachable_cache is not None:
        return _db_reachable_cache
    from sqlalchemy import text

    try:
        async with asyncio.timeout(3):
            async with get_engine().connect() as conn:
                await conn.execute(text("SELECT 1"))
        _db_reachable_cache = True
    except Exception:
        _db_reachable_cache = False
    return _db_reachable_cache


async def _seed_album_with_cover(*, with_cover: bool) -> tuple[uuid.UUID, uuid.UUID]:
    """Create an artist + album (+ optional cover file) + track. Returns
    ``(album_id, track_id)``."""
    async with session_scope() as session:
        artist = await SqlAlchemyArtistRepository(session).get_or_create("Coverart Artist")
        album = await SqlAlchemyAlbumRepository(session).get_or_create(
            title="Coverart Album", artist_id=artist.id, year=2020, musicbrainz_id=None
        )
        track = await SqlAlchemyTrackRepository(session).add(
            id=uuid.uuid4(),
            title="Coverart Track",
            artist_id=artist.id,
            storage_uri="tracks/zz/cover-track.mp3",
            file_format="mp3",
            file_size=10,
            source="upload",
            source_id="cover-src",
            metadata_status="enriched",
            added_by=None,
        )
        # Link the track to the album (add() doesn't take album_id).
        await SqlAlchemyTrackRepository(session).apply_enrichment(
            track.id,
            title="Coverart Track",
            artist_id=artist.id,
            album_id=album.id,
            genre=None,
            year=2020,
            track_number=1,
            duration_seconds=1,
            bitrate=None,
            acoustid_fingerprint=None,
            musicbrainz_id=None,
            metadata_status="enriched",
        )
        if with_cover:
            key = f"covers/{album.id}.png"
            import tempfile

            with tempfile.NamedTemporaryFile(suffix=".png") as tmp:
                tmp.write(_PNG_BYTES)
                tmp.flush()
                await get_file_storage().save_file(key, Path(tmp.name))
            await SqlAlchemyAlbumRepository(session).set_cover_path(album.id, key)
        return album.id, track.id


@pytest.fixture
async def api(tmp_path: Path) -> AsyncIterator[AsyncClient]:
    if not await _db_reachable():
        pytest.skip("Postgres not reachable — integration test skipped.")

    os.environ["MEDIA_PATH"] = str(tmp_path)
    get_settings.cache_clear()

    import app.infrastructure.storage.provider as _storage_provider

    _storage_provider._storage = None

    try:
        async with get_engine().begin() as conn:
            await conn.run_sync(Base.metadata.drop_all)
            await conn.run_sync(Base.metadata.create_all)

        from app.application.user_service import UserService
        from app.core.security import Argon2PasswordHasher

        async with session_scope() as session:
            await UserService(
                users=SqlAlchemyUserRepository(session),
                refresh_tokens=SqlAlchemyRefreshTokenRepository(session),
                hasher=Argon2PasswordHasher(),
            ).create_user(username="testuser", password="testpass1", is_superuser=False)

        from app.main import create_app

        app = create_app()
        async with LifespanManager(app):
            transport = ASGITransport(app=app)
            async with AsyncClient(transport=transport, base_url="http://test") as client:
                yield client

        async with get_engine().begin() as conn:
            await conn.run_sync(Base.metadata.drop_all)
        await dispose_engine()
    finally:
        _storage_provider._storage = None
        os.environ.pop("MEDIA_PATH", None)
        get_settings.cache_clear()


async def _login(api: AsyncClient) -> str:
    resp = await api.post(
        "/api/v1/auth/login", json={"username": "testuser", "password": "testpass1"}
    )
    assert resp.status_code == 200
    return str(resp.json()["access_token"])


async def test_album_cover_served(api: AsyncClient) -> None:
    token = await _login(api)
    album_id, _ = await _seed_album_with_cover(with_cover=True)

    resp = await api.get(f"/api/v1/albums/{album_id}/cover?token={token}")
    assert resp.status_code == 200, resp.text
    assert resp.headers["content-type"] == "image/png"
    assert resp.content == _PNG_BYTES


async def test_track_cover_served_from_album(api: AsyncClient) -> None:
    token = await _login(api)
    _, track_id = await _seed_album_with_cover(with_cover=True)

    resp = await api.get(f"/api/v1/tracks/{track_id}/cover?token={token}")
    assert resp.status_code == 200, resp.text
    assert resp.headers["content-type"] == "image/png"
    assert resp.content == _PNG_BYTES


async def test_album_without_cover_is_404(api: AsyncClient) -> None:
    token = await _login(api)
    album_id, _ = await _seed_album_with_cover(with_cover=False)

    resp = await api.get(f"/api/v1/albums/{album_id}/cover?token={token}")
    assert resp.status_code == 404


async def test_cover_requires_auth(api: AsyncClient) -> None:
    album_id, _ = await _seed_album_with_cover(with_cover=True)
    resp = await api.get(f"/api/v1/albums/{album_id}/cover")
    assert resp.status_code == 401


async def test_album_appears_with_has_cover_flag(api: AsyncClient) -> None:
    token = await _login(api)
    album_id, _ = await _seed_album_with_cover(with_cover=True)

    resp = await api.get(f"/api/v1/albums/{album_id}", headers={"Authorization": f"Bearer {token}"})
    assert resp.status_code == 200
    assert resp.json()["has_cover"] is True