2026-03-12 01:40:04 -04:00
|
|
|
|
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()
|
2026-03-12 02:09:15 -04:00
|
|
|
|
assert "episode_number" in data
|
2026-03-12 01:40:04 -04:00
|
|
|
|
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
|
2026-03-12 02:09:15 -04:00
|
|
|
|
data = resp.json()
|
|
|
|
|
|
assert len(data) >= 1
|
|
|
|
|
|
assert "episode_number" in data[0]
|
2026-03-12 01:40:04 -04:00
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_shows_detail(client, db):
|
|
|
|
|
|
show = _seed_show(db)
|
|
|
|
|
|
resp = client.get(f"/shows/{show.id}")
|
|
|
|
|
|
assert resp.status_code == 200
|
2026-03-12 02:09:15 -04:00
|
|
|
|
data = resp.json()
|
|
|
|
|
|
assert "episode_number" in data
|
|
|
|
|
|
assert len(data["tracks"]) == 2
|
2026-03-12 01:40:04 -04:00
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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
|