"""Integration tests for the storage maintenance endpoints. Covers ``/storage/duplicates`` (shared fingerprint), ``/broken`` (failed enrichment), ``/missing-metadata`` (pending), and the admin-gated ``/cleanup`` enqueue. 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 ( SqlAlchemyArtistRepository, SqlAlchemyRefreshTokenRepository, SqlAlchemyTrackRepository, SqlAlchemyUserRepository, ) from asgi_lifespan import LifespanManager from httpx import ASGITransport, AsyncClient pytestmark = pytest.mark.asyncio _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_track( *, title: str, source_id: str, metadata_status: str, fingerprint: str | None ) -> uuid.UUID: async with session_scope() as session: artist = await SqlAlchemyArtistRepository(session).get_or_create("Storage Artist") tracks = SqlAlchemyTrackRepository(session) tid = uuid.uuid4() await tracks.add( id=tid, title=title, artist_id=artist.id, storage_uri=f"tracks/zz/{source_id}.mp3", file_format="mp3", file_size=10, source="upload", source_id=source_id, metadata_status="pending", added_by=None, ) # apply_enrichment lets us set fingerprint + final status precisely. await tracks.apply_enrichment( tid, title=title, artist_id=artist.id, album_id=None, genre=None, year=None, track_number=None, duration_seconds=1, bitrate=None, acoustid_fingerprint=fingerprint, musicbrainz_id=None, metadata_status=metadata_status, ) return tid @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() 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: svc = UserService( users=SqlAlchemyUserRepository(session), refresh_tokens=SqlAlchemyRefreshTokenRepository(session), hasher=Argon2PasswordHasher(), ) await svc.create_user(username="user", password="testpass1", is_superuser=False) await svc.create_user(username="admin", password="testpass1", is_superuser=True) 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: os.environ.pop("MEDIA_PATH", None) get_settings.cache_clear() async def _auth(api: AsyncClient, username: str) -> dict[str, str]: resp = await api.post( "/api/v1/auth/login", json={"username": username, "password": "testpass1"} ) assert resp.status_code == 200, resp.text return {"Authorization": f"Bearer {resp.json()['access_token']}"} async def test_duplicates_grouped_by_fingerprint(api: AsyncClient) -> None: headers = await _auth(api, "user") await _seed_track(title="Dup A", source_id="a", metadata_status="enriched", fingerprint="FP1") await _seed_track(title="Dup B", source_id="b", metadata_status="enriched", fingerprint="FP1") # A lone fingerprint must NOT show up as a duplicate. await _seed_track(title="Solo", source_id="c", metadata_status="enriched", fingerprint="FP2") resp = await api.get("/api/v1/storage/duplicates", headers=headers) assert resp.status_code == 200, resp.text groups = resp.json() assert len(groups) == 1 assert groups[0]["fingerprint"] == "FP1" assert len(groups[0]["tracks"]) == 2 async def test_broken_lists_failed_tracks(api: AsyncClient) -> None: headers = await _auth(api, "user") await _seed_track(title="Bad", source_id="x", metadata_status="failed", fingerprint=None) await _seed_track(title="Good", source_id="y", metadata_status="enriched", fingerprint=None) resp = await api.get("/api/v1/storage/broken", headers=headers) assert resp.status_code == 200, resp.text body = resp.json() assert body["total"] == 1 assert body["items"][0]["title"] == "Bad" async def test_missing_metadata_lists_pending(api: AsyncClient) -> None: headers = await _auth(api, "user") await _seed_track(title="Pending", source_id="p", metadata_status="pending", fingerprint=None) await _seed_track(title="Done", source_id="d", metadata_status="enriched", fingerprint=None) resp = await api.get("/api/v1/storage/missing-metadata", headers=headers) assert resp.status_code == 200, resp.text body = resp.json() assert body["total"] == 1 assert body["items"][0]["title"] == "Pending" async def test_cleanup_requires_admin(api: AsyncClient) -> None: user = await _auth(api, "user") resp = await api.post("/api/v1/storage/cleanup", headers=user) assert resp.status_code == 403 async def test_cleanup_enqueues_for_admin(api: AsyncClient) -> None: admin = await _auth(api, "admin") resp = await api.post("/api/v1/storage/cleanup", headers=admin) # 202 with a job id when the queue is reachable; 503 if Redis is down. assert resp.status_code in (202, 503), resp.text if resp.status_code == 202: assert resp.json()["status"] == "enqueued" assert resp.json()["job_id"]