Senko-san
316 lines
11 KiB
Python
316 lines
11 KiB
Python
"""Integration tests for the v1 playlist API.
|
|
|
|
Requires a reachable Postgres; skips otherwise.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
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 (
|
|
SqlAlchemyRefreshTokenRepository,
|
|
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
|
|
|
|
|
|
@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:
|
|
service = UserService(
|
|
users=SqlAlchemyUserRepository(session),
|
|
refresh_tokens=SqlAlchemyRefreshTokenRepository(session),
|
|
hasher=Argon2PasswordHasher(),
|
|
)
|
|
await service.create_user(username="testuser", password="testpass1", is_superuser=False)
|
|
await service.create_user(
|
|
username="otheruser", password="otherpass1", 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, *, username: str = "testuser", password: str = "testpass1"
|
|
) -> str:
|
|
resp = await api.post("/api/v1/auth/login", json={"username": username, "password": password})
|
|
assert resp.status_code == 200
|
|
return str(resp.json()["access_token"])
|
|
|
|
|
|
async def _upload(api: AsyncClient, token: str, *, name: str = "song.mp3") -> str:
|
|
# Content must vary per call: uploads dedupe by content hash, so identical
|
|
# bytes across "different" tracks would collapse to a single track_id.
|
|
audio = f"fake audio bytes for playlist test {name}".encode() * 10
|
|
resp = await api.post(
|
|
"/api/v1/upload",
|
|
files={"file": (name, audio, "audio/mpeg")},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
return str(resp.json()["track_id"])
|
|
|
|
|
|
async def _create_playlist(
|
|
api: AsyncClient, token: str, *, name: str = "My Playlist"
|
|
) -> dict[str, Any]:
|
|
resp = await api.post(
|
|
"/api/v1/playlists",
|
|
json={"name": name, "description": "desc"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
body: dict[str, Any] = resp.json()
|
|
return body
|
|
|
|
|
|
async def test_playlist_lifecycle(api: AsyncClient) -> None:
|
|
token = await _login(api)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
created = await _create_playlist(api, token)
|
|
playlist_id = created["id"]
|
|
assert created["version"] == 1
|
|
assert created["track_count"] == 0
|
|
|
|
resp = await api.get("/api/v1/playlists", headers=headers)
|
|
assert resp.status_code == 200, resp.text
|
|
assert any(p["id"] == playlist_id for p in resp.json()["items"])
|
|
|
|
resp = await api.get(f"/api/v1/playlists/{playlist_id}", headers=headers)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["name"] == "My Playlist"
|
|
|
|
resp = await api.patch(
|
|
f"/api/v1/playlists/{playlist_id}",
|
|
json={"name": "Renamed", "description": "new desc"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["name"] == "Renamed"
|
|
assert body["description"] == "new desc"
|
|
assert body["version"] == 2
|
|
|
|
resp = await api.delete(f"/api/v1/playlists/{playlist_id}", headers=headers)
|
|
assert resp.status_code == 204
|
|
|
|
resp = await api.get(f"/api/v1/playlists/{playlist_id}", headers=headers)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
async def test_add_track_idempotent(api: AsyncClient) -> None:
|
|
token = await _login(api)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
playlist_id = (await _create_playlist(api, token))["id"]
|
|
track_id = await _upload(api, token)
|
|
|
|
resp = await api.post(
|
|
f"/api/v1/playlists/{playlist_id}/tracks", json={"track_id": track_id}, headers=headers
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
resp = await api.get(f"/api/v1/playlists/{playlist_id}", headers=headers)
|
|
assert resp.json()["version"] == 2
|
|
assert resp.json()["track_count"] == 1
|
|
|
|
resp = await api.get(f"/api/v1/playlists/{playlist_id}/tracks", headers=headers)
|
|
assert resp.status_code == 200, resp.text
|
|
assert [t["id"] for t in resp.json()["items"]] == [track_id]
|
|
|
|
# Adding the same track again must be idempotent: 204, no duplicate, no version bump.
|
|
resp = await api.post(
|
|
f"/api/v1/playlists/{playlist_id}/tracks", json={"track_id": track_id}, headers=headers
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
resp = await api.get(f"/api/v1/playlists/{playlist_id}", headers=headers)
|
|
assert resp.json()["version"] == 2
|
|
assert resp.json()["track_count"] == 1
|
|
|
|
|
|
async def test_remove_track(api: AsyncClient) -> None:
|
|
token = await _login(api)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
playlist_id = (await _create_playlist(api, token))["id"]
|
|
track_id = await _upload(api, token)
|
|
|
|
resp = await api.post(
|
|
f"/api/v1/playlists/{playlist_id}/tracks", json={"track_id": track_id}, headers=headers
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
resp = await api.delete(f"/api/v1/playlists/{playlist_id}/tracks/{track_id}", headers=headers)
|
|
assert resp.status_code == 204
|
|
|
|
resp = await api.get(f"/api/v1/playlists/{playlist_id}", headers=headers)
|
|
assert resp.json()["track_count"] == 0
|
|
assert resp.json()["version"] == 3
|
|
|
|
|
|
async def test_reorder_happy_path(api: AsyncClient) -> None:
|
|
token = await _login(api)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
playlist_id = (await _create_playlist(api, token))["id"]
|
|
|
|
track_ids = []
|
|
for i in range(3):
|
|
track_id = await _upload(api, token, name=f"song{i}.mp3")
|
|
track_ids.append(track_id)
|
|
resp = await api.post(
|
|
f"/api/v1/playlists/{playlist_id}/tracks",
|
|
json={"track_id": track_id},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
reordered = list(reversed(track_ids))
|
|
resp = await api.put(
|
|
f"/api/v1/playlists/{playlist_id}/tracks/reorder",
|
|
json={"track_ids": reordered},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["version"] == 5 # 1 create + 3 adds + 1 reorder
|
|
|
|
resp = await api.get(f"/api/v1/playlists/{playlist_id}/tracks", headers=headers)
|
|
assert resp.status_code == 200, resp.text
|
|
assert [t["id"] for t in resp.json()["items"]] == reordered
|
|
|
|
|
|
async def test_reorder_bad_id_set_returns_422(api: AsyncClient) -> None:
|
|
token = await _login(api)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
playlist_id = (await _create_playlist(api, token))["id"]
|
|
track_id = await _upload(api, token)
|
|
|
|
resp = await api.post(
|
|
f"/api/v1/playlists/{playlist_id}/tracks", json={"track_id": track_id}, headers=headers
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# Missing the only member.
|
|
resp = await api.put(
|
|
f"/api/v1/playlists/{playlist_id}/tracks/reorder",
|
|
json={"track_ids": []},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
# Duplicate id.
|
|
resp = await api.put(
|
|
f"/api/v1/playlists/{playlist_id}/tracks/reorder",
|
|
json={"track_ids": [track_id, track_id]},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
# Extra unknown id.
|
|
other_id = await _upload(api, token, name="not-a-member.mp3")
|
|
resp = await api.put(
|
|
f"/api/v1/playlists/{playlist_id}/tracks/reorder",
|
|
json={"track_ids": [track_id, other_id]},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
async def test_reorder_by_non_owner_returns_403(api: AsyncClient) -> None:
|
|
token = await _login(api)
|
|
other_token = await _login(api, username="otheruser", password="otherpass1")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
playlist_id = (await _create_playlist(api, token))["id"]
|
|
track_id = await _upload(api, token)
|
|
|
|
resp = await api.post(
|
|
f"/api/v1/playlists/{playlist_id}/tracks", json={"track_id": track_id}, headers=headers
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
resp = await api.put(
|
|
f"/api/v1/playlists/{playlist_id}/tracks/reorder",
|
|
json={"track_ids": [track_id]},
|
|
headers={"Authorization": f"Bearer {other_token}"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_unknown_playlist_returns_404(api: AsyncClient) -> None:
|
|
token = await _login(api)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
unknown_id = "00000000-0000-0000-0000-000000000000"
|
|
track_id = await _upload(api, token)
|
|
|
|
resp = await api.get(f"/api/v1/playlists/{unknown_id}", headers=headers)
|
|
assert resp.status_code == 404
|
|
|
|
resp = await api.patch(f"/api/v1/playlists/{unknown_id}", json={"name": "x"}, headers=headers)
|
|
assert resp.status_code == 404
|
|
|
|
resp = await api.delete(f"/api/v1/playlists/{unknown_id}", headers=headers)
|
|
assert resp.status_code == 404
|
|
|
|
resp = await api.post(
|
|
f"/api/v1/playlists/{unknown_id}/tracks", json={"track_id": track_id}, headers=headers
|
|
)
|
|
assert resp.status_code == 404
|