92136f0508
- Add web_user, web_password, secret_key as optional params to create_app - Conditionally mount dashboard router when all three are set - Pass settings from main.py to create_app - Add tests for no dashboard/login routes when config absent Made-with: Cursor
155 lines
4.5 KiB
Python
155 lines
4.5 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
|