425a7047c3
Public-facing page at / shows the current show's playlist with tracks obscured until the admin marks them as announced. Tracks reveal in real-time via a new unauthenticated /ws/public WebSocket. Server-side censorship on /public/playlist strips track details from unannounced items and sanitizes announced tracks to only expose frontend-needed fields (no raw_json, track_id, etc). Past episodes are browsable with fully revealed but sanitized tracklists. Also fixes RuntimeError on backfill shutdown by closing the httpx client on the same event loop that created it. Made-with: Cursor
224 lines
6.4 KiB
Python
224 lines
6.4 KiB
Python
from datetime import datetime, timezone
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from ntr_fetcher.api import create_app
|
||
from ntr_fetcher.db import Database
|
||
from ntr_fetcher.models import Track
|
||
|
||
# Fixed "now" so seeded show (week_start Mar 12 2am–Mar 19 2am UTC) matches get_show_week
|
||
FAKE_NOW = datetime(2026, 3, 14, 12, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
@pytest.fixture
|
||
def db(tmp_path):
|
||
database = Database(str(tmp_path / "test.db"))
|
||
database.initialize()
|
||
return database
|
||
|
||
|
||
@pytest.fixture
|
||
def app(db):
|
||
poller = MagicMock()
|
||
poller.last_fetch = datetime(2026, 3, 12, 12, 0, 0, tzinfo=timezone.utc)
|
||
poller.alive = True
|
||
poller.poll_once = AsyncMock()
|
||
return create_app(db=db, poller=poller, admin_token="test-token")
|
||
|
||
|
||
@pytest.fixture
|
||
def client(app):
|
||
with patch("ntr_fetcher.api.datetime") as mock_dt:
|
||
mock_dt.now.return_value = FAKE_NOW
|
||
yield TestClient(app)
|
||
|
||
|
||
def _seed_show(db):
|
||
week_start = datetime(2026, 3, 12, 2, 0, 0, tzinfo=timezone.utc)
|
||
week_end = datetime(2026, 3, 19, 2, 0, 0, tzinfo=timezone.utc)
|
||
show = db.get_or_create_show(week_start, week_end)
|
||
t1 = Track(1, "Song A", "Artist A", "https://soundcloud.com/a/1", None, 180000, "cc-by",
|
||
datetime(2026, 3, 14, 1, 0, 0, tzinfo=timezone.utc), "{}")
|
||
t2 = Track(2, "Song B", "Artist B", "https://soundcloud.com/b/2", None, 200000, "cc-by-sa",
|
||
datetime(2026, 3, 14, 2, 0, 0, tzinfo=timezone.utc), "{}")
|
||
db.upsert_track(t1)
|
||
db.upsert_track(t2)
|
||
db.set_show_tracks(show.id, [t1.id, t2.id])
|
||
return show
|
||
|
||
|
||
def test_health(client):
|
||
resp = client.get("/health")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["status"] == "ok"
|
||
assert data["poller_alive"] is True
|
||
|
||
|
||
def test_playlist(client, db):
|
||
_seed_show(db)
|
||
resp = client.get("/playlist")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "episode_number" in data
|
||
assert len(data["tracks"]) == 2
|
||
assert data["tracks"][0]["position"] == 1
|
||
assert data["tracks"][0]["title"] == "Song A"
|
||
|
||
|
||
def test_playlist_by_position(client, db):
|
||
_seed_show(db)
|
||
resp = client.get("/playlist/2")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["title"] == "Song B"
|
||
|
||
|
||
def test_playlist_by_position_not_found(client, db):
|
||
_seed_show(db)
|
||
resp = client.get("/playlist/99")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
def test_shows_list(client, db):
|
||
_seed_show(db)
|
||
resp = client.get("/shows")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data) >= 1
|
||
assert "episode_number" in data[0]
|
||
|
||
|
||
def test_shows_detail(client, db):
|
||
show = _seed_show(db)
|
||
resp = client.get(f"/shows/{show.id}")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "episode_number" in data
|
||
assert len(data["tracks"]) == 2
|
||
|
||
|
||
def test_admin_refresh_requires_token(client):
|
||
resp = client.post("/admin/refresh")
|
||
assert resp.status_code == 401
|
||
|
||
|
||
def test_admin_refresh_with_token(client):
|
||
resp = client.post(
|
||
"/admin/refresh",
|
||
headers={"Authorization": "Bearer test-token"},
|
||
json={"full": False},
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
|
||
def test_admin_remove_track(client, db):
|
||
show = _seed_show(db)
|
||
resp = client.delete(
|
||
"/admin/tracks/1",
|
||
headers={"Authorization": "Bearer test-token"},
|
||
)
|
||
assert resp.status_code == 200
|
||
tracks = db.get_show_tracks(show.id)
|
||
assert len(tracks) == 1
|
||
|
||
|
||
def test_show_by_episode(client, db):
|
||
week_start = datetime(2026, 3, 12, 2, 0, 0, tzinfo=timezone.utc)
|
||
week_end = datetime(2026, 3, 19, 2, 0, 0, tzinfo=timezone.utc)
|
||
show = db.get_or_create_show(week_start, week_end, episode_number=530)
|
||
t1 = Track(1, "Song A", "Artist A", "https://soundcloud.com/a/1", None, 180000, "cc-by",
|
||
datetime(2026, 3, 14, 1, 0, 0, tzinfo=timezone.utc), "{}")
|
||
db.upsert_track(t1)
|
||
db.set_show_tracks(show.id, [t1.id])
|
||
resp = client.get("/shows/by-episode/530")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["episode_number"] == 530
|
||
assert len(data["tracks"]) == 1
|
||
|
||
|
||
def test_show_by_episode_not_found(client):
|
||
resp = client.get("/shows/by-episode/999")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
def test_no_dashboard_routes_without_config(client):
|
||
resp = client.get("/dashboard", follow_redirects=False)
|
||
assert resp.status_code == 404
|
||
|
||
|
||
def test_no_login_route_without_config(client):
|
||
resp = client.get("/login")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
# --- Public endpoint tests ---
|
||
|
||
|
||
def test_index_page(client):
|
||
resp = client.get("/")
|
||
assert resp.status_code == 200
|
||
assert "NtR Playlist" in resp.text
|
||
|
||
|
||
def test_public_playlist_censors_unannounced(client, db):
|
||
show = _seed_show(db)
|
||
db.set_track_announced(show.id, 1, True)
|
||
|
||
resp = client.get("/public/playlist")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data["tracks"]) == 2
|
||
|
||
revealed = data["tracks"][0]
|
||
assert revealed["announced"] == 1
|
||
assert revealed["title"] == "Song A"
|
||
assert revealed["artist"] == "Artist A"
|
||
assert "raw_json" not in revealed
|
||
assert "track_id" not in revealed
|
||
assert "show_id" not in revealed
|
||
|
||
hidden = data["tracks"][1]
|
||
assert hidden["announced"] == 0
|
||
assert hidden["position"] == 2
|
||
assert "title" not in hidden
|
||
assert "artist" not in hidden
|
||
|
||
|
||
def test_public_playlist_all_hidden(client, db):
|
||
_seed_show(db)
|
||
resp = client.get("/public/playlist")
|
||
data = resp.json()
|
||
for t in data["tracks"]:
|
||
assert t["announced"] == 0
|
||
assert "title" not in t
|
||
|
||
|
||
def test_public_shows_list(client, db):
|
||
_seed_show(db)
|
||
resp = client.get("/public/shows")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data) >= 1
|
||
assert "id" in data[0]
|
||
assert "episode_number" in data[0]
|
||
|
||
|
||
def test_public_show_detail_fully_revealed(client, db):
|
||
show = _seed_show(db)
|
||
resp = client.get(f"/public/shows/{show.id}")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data["tracks"]) == 2
|
||
assert data["tracks"][0]["title"] == "Song A"
|
||
assert data["tracks"][1]["title"] == "Song B"
|
||
assert "raw_json" not in data["tracks"][0]
|
||
assert "track_id" not in data["tracks"][0]
|
||
|
||
|
||
def test_public_show_detail_not_found(client):
|
||
resp = client.get("/public/shows/999")
|
||
assert resp.status_code == 404
|