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
315 lines
9.2 KiB
Python
315 lines
9.2 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from ntr_fetcher.dashboard import create_dashboard_router
|
|
from ntr_fetcher.db import Database
|
|
from ntr_fetcher.models import Track
|
|
from ntr_fetcher.websocket import AnnounceManager, PublicManager
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
database = Database(str(tmp_path / "test.db"))
|
|
database.initialize()
|
|
return database
|
|
|
|
|
|
@pytest.fixture
|
|
def manager():
|
|
return AnnounceManager()
|
|
|
|
|
|
@pytest.fixture
|
|
def public_manager():
|
|
return PublicManager()
|
|
|
|
|
|
@pytest.fixture
|
|
def app(db, manager, public_manager):
|
|
a = FastAPI()
|
|
router = create_dashboard_router(
|
|
db=db,
|
|
manager=manager,
|
|
public_manager=public_manager,
|
|
admin_token="test-token",
|
|
web_user="nick",
|
|
web_password="secret",
|
|
secret_key="test-secret-key",
|
|
show_day=2,
|
|
show_hour=22,
|
|
)
|
|
a.include_router(router)
|
|
return a
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
return 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), "{}")
|
|
db.upsert_track(t1)
|
|
db.set_show_tracks(show.id, [t1.id])
|
|
return show
|
|
|
|
|
|
# --- Session auth tests ---
|
|
|
|
def test_dashboard_redirects_without_session(client):
|
|
resp = client.get("/dashboard", follow_redirects=False)
|
|
assert resp.status_code == 303
|
|
assert "/login" in resp.headers["location"]
|
|
|
|
|
|
def test_login_page_renders(client):
|
|
resp = client.get("/login")
|
|
assert resp.status_code == 200
|
|
assert "login" in resp.text.lower()
|
|
|
|
|
|
def test_login_with_valid_credentials(client):
|
|
resp = client.post(
|
|
"/login",
|
|
data={"username": "nick", "password": "secret"},
|
|
follow_redirects=False,
|
|
)
|
|
assert resp.status_code == 303
|
|
assert "/dashboard" in resp.headers["location"]
|
|
assert "ntr_session" in resp.cookies
|
|
|
|
|
|
def test_login_with_invalid_credentials(client):
|
|
resp = client.post(
|
|
"/login",
|
|
data={"username": "nick", "password": "wrong"},
|
|
follow_redirects=False,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "invalid" in resp.text.lower() or "incorrect" in resp.text.lower()
|
|
|
|
|
|
def test_dashboard_accessible_with_session(client):
|
|
client.post(
|
|
"/login",
|
|
data={"username": "nick", "password": "secret"},
|
|
follow_redirects=False,
|
|
)
|
|
resp = client.get("/dashboard")
|
|
assert resp.status_code == 200
|
|
assert "dashboard" in resp.text.lower()
|
|
|
|
|
|
def test_logout_clears_session(client):
|
|
client.post(
|
|
"/login",
|
|
data={"username": "nick", "password": "secret"},
|
|
follow_redirects=False,
|
|
)
|
|
resp = client.get("/logout", follow_redirects=False)
|
|
assert resp.status_code == 303
|
|
assert "/login" in resp.headers["location"]
|
|
|
|
resp2 = client.get("/dashboard", follow_redirects=False)
|
|
assert resp2.status_code == 303
|
|
|
|
|
|
# --- Announce endpoint tests ---
|
|
|
|
def test_announce_with_session(client, db):
|
|
_seed_show(db)
|
|
client.post(
|
|
"/login",
|
|
data={"username": "nick", "password": "secret"},
|
|
follow_redirects=False,
|
|
)
|
|
resp = client.post("/admin/announce", json={"show_id": 1, "position": 1})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "announced"
|
|
assert "Now Playing:" in data["message"]
|
|
assert "Song A" in data["message"]
|
|
assert "Artist A" in data["message"]
|
|
|
|
|
|
def test_announce_with_bearer(client, db):
|
|
_seed_show(db)
|
|
resp = client.post(
|
|
"/admin/announce",
|
|
json={"show_id": 1, "position": 1},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "Now Playing:" in resp.json()["message"]
|
|
|
|
|
|
def test_announce_without_auth(client, db):
|
|
_seed_show(db)
|
|
resp = client.post("/admin/announce", json={"show_id": 1, "position": 1})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_announce_invalid_position(client, db):
|
|
_seed_show(db)
|
|
resp = client.post(
|
|
"/admin/announce",
|
|
json={"show_id": 1, "position": 99},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# --- WebSocket tests ---
|
|
|
|
def test_ws_subscribe_bot_with_valid_token(app):
|
|
with TestClient(app) as c:
|
|
with c.websocket_connect("/ws/announce") as ws:
|
|
ws.send_json({"type": "subscribe", "token": "test-token", "role": "bot", "client_id": "test-bot"})
|
|
data = ws.receive_json()
|
|
assert data["type"] == "status"
|
|
assert data["subscribers"] == 1
|
|
assert data["clients"][0]["client_id"] == "test-bot"
|
|
|
|
|
|
def test_ws_subscribe_viewer_not_counted(app):
|
|
with TestClient(app) as c:
|
|
with c.websocket_connect("/ws/announce") as ws:
|
|
ws.send_json({"type": "subscribe", "token": "test-token", "role": "viewer"})
|
|
data = ws.receive_json()
|
|
assert data["type"] == "status"
|
|
assert data["subscribers"] == 0
|
|
|
|
|
|
def test_ws_subscribe_with_invalid_token(app):
|
|
with TestClient(app) as c:
|
|
with c.websocket_connect("/ws/announce") as ws:
|
|
ws.send_json({"type": "subscribe", "token": "wrong"})
|
|
with pytest.raises(Exception):
|
|
ws.receive_json()
|
|
|
|
|
|
# --- Announced endpoint tests ---
|
|
|
|
def test_announce_sets_announced_flag(client, db):
|
|
show = _seed_show(db)
|
|
resp = client.post(
|
|
"/admin/announce",
|
|
json={"show_id": show.id, "position": 1},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert resp.status_code == 200
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert tracks[0]["announced"] == 1
|
|
|
|
|
|
def test_set_announced_with_bearer(client, db):
|
|
show = _seed_show(db)
|
|
resp = client.post(
|
|
"/admin/announced",
|
|
json={"show_id": show.id, "position": 1, "announced": True},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert tracks[0]["announced"] == 1
|
|
|
|
|
|
def test_set_announced_toggle_off(client, db):
|
|
show = _seed_show(db)
|
|
client.post(
|
|
"/admin/announced",
|
|
json={"show_id": show.id, "position": 1, "announced": True},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
resp = client.post(
|
|
"/admin/announced",
|
|
json={"show_id": show.id, "position": 1, "announced": False},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert resp.status_code == 200
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert tracks[0]["announced"] == 0
|
|
|
|
|
|
def test_set_announced_without_auth(client, db):
|
|
_seed_show(db)
|
|
resp = client.post(
|
|
"/admin/announced",
|
|
json={"show_id": 1, "position": 1, "announced": True},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_set_announced_invalid_position(client, db):
|
|
_seed_show(db)
|
|
resp = client.post(
|
|
"/admin/announced",
|
|
json={"show_id": 1, "position": 99, "announced": True},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# --- Public WebSocket tests ---
|
|
|
|
|
|
def test_ws_public_connects_without_auth(app):
|
|
with TestClient(app) as c:
|
|
with c.websocket_connect("/ws/public") as ws:
|
|
pass
|
|
|
|
|
|
def test_ws_public_receives_reveal_on_announce(app, db, public_manager):
|
|
_seed_show(db)
|
|
with TestClient(app) as c:
|
|
with c.websocket_connect("/ws/public") as ws:
|
|
c.post(
|
|
"/admin/announce",
|
|
json={"show_id": 1, "position": 1},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
msg = ws.receive_json()
|
|
assert msg["type"] == "reveal"
|
|
assert msg["position"] == 1
|
|
assert msg["track"]["title"] == "Song A"
|
|
assert msg["track"]["artist"] == "Artist A"
|
|
|
|
|
|
def test_ws_public_receives_reveal_on_announced_toggle(app, db, public_manager):
|
|
_seed_show(db)
|
|
with TestClient(app) as c:
|
|
with c.websocket_connect("/ws/public") as ws:
|
|
c.post(
|
|
"/admin/announced",
|
|
json={"show_id": 1, "position": 1, "announced": True},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
msg = ws.receive_json()
|
|
assert msg["type"] == "reveal"
|
|
assert msg["position"] == 1
|
|
assert msg["track"]["title"] == "Song A"
|
|
|
|
|
|
def test_ws_public_receives_hide_on_unannounce(app, db, public_manager):
|
|
show = _seed_show(db)
|
|
db.set_track_announced(show.id, 1, True)
|
|
with TestClient(app) as c:
|
|
with c.websocket_connect("/ws/public") as ws:
|
|
c.post(
|
|
"/admin/announced",
|
|
json={"show_id": 1, "position": 1, "announced": False},
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
msg = ws.receive_json()
|
|
assert msg["type"] == "hide"
|
|
assert msg["position"] == 1
|
|
assert "track" not in msg
|