1d08580a45
Allows looking up shows by episode number instead of internal DB ID, enabling IRC bot commands like !playlist 530 to resolve directly. Made-with: Cursor
298 lines
11 KiB
Python
298 lines
11 KiB
Python
import sqlite3
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from ntr_fetcher.db import Database
|
|
from ntr_fetcher.models import Track
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
db_path = str(tmp_path / "test.db")
|
|
database = Database(db_path)
|
|
database.initialize()
|
|
return database
|
|
|
|
|
|
def test_tables_created(db):
|
|
conn = sqlite3.connect(db.path)
|
|
cursor = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
|
)
|
|
tables = [row[0] for row in cursor.fetchall()]
|
|
conn.close()
|
|
assert "tracks" in tables
|
|
assert "shows" in tables
|
|
assert "show_tracks" in tables
|
|
|
|
|
|
def test_initialize_idempotent(db):
|
|
"""Calling initialize twice doesn't raise."""
|
|
db.initialize()
|
|
|
|
|
|
def _make_track(id: int, liked_at: str, title: str = "Test", artist: str = "Artist") -> Track:
|
|
return Track(
|
|
id=id,
|
|
title=title,
|
|
artist=artist,
|
|
permalink_url=f"https://soundcloud.com/test/track-{id}",
|
|
artwork_url=None,
|
|
duration_ms=180000,
|
|
license="cc-by",
|
|
liked_at=datetime.fromisoformat(liked_at),
|
|
raw_json="{}",
|
|
)
|
|
|
|
|
|
def test_upsert_track(db):
|
|
track = _make_track(100, "2026-03-10T12:00:00+00:00")
|
|
db.upsert_track(track)
|
|
result = db.get_track(100)
|
|
assert result is not None
|
|
assert result.title == "Test"
|
|
|
|
|
|
def test_upsert_track_updates_existing(db):
|
|
track1 = _make_track(100, "2026-03-10T12:00:00+00:00", title="Original")
|
|
db.upsert_track(track1)
|
|
track2 = _make_track(100, "2026-03-10T12:00:00+00:00", title="Updated")
|
|
db.upsert_track(track2)
|
|
result = db.get_track(100)
|
|
assert result.title == "Updated"
|
|
|
|
|
|
def test_get_or_create_show(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
assert show.id is not None
|
|
assert show.week_start == week_start
|
|
show2 = db.get_or_create_show(week_start, week_end)
|
|
assert show2.id == show.id
|
|
|
|
|
|
def test_set_show_tracks(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00", title="First")
|
|
t2 = _make_track(2, "2026-03-14T02:00:00+00:00", title="Second")
|
|
db.upsert_track(t1)
|
|
db.upsert_track(t2)
|
|
db.set_show_tracks(show.id, [t1.id, t2.id])
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert len(tracks) == 2
|
|
assert tracks[0]["position"] == 1
|
|
assert tracks[0]["title"] == "First"
|
|
assert tracks[1]["position"] == 2
|
|
|
|
|
|
def test_set_show_tracks_preserves_existing_positions(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00")
|
|
db.upsert_track(t1)
|
|
db.set_show_tracks(show.id, [t1.id])
|
|
t2 = _make_track(2, "2026-03-14T02:00:00+00:00")
|
|
db.upsert_track(t2)
|
|
db.set_show_tracks(show.id, [t1.id, t2.id])
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert tracks[0]["track_id"] == 1
|
|
assert tracks[0]["position"] == 1
|
|
assert tracks[1]["track_id"] == 2
|
|
assert tracks[1]["position"] == 2
|
|
|
|
|
|
def test_set_show_tracks_removes_unliked(db):
|
|
"""Tracks no longer in the likes list are removed and positions re-compact."""
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00", title="First")
|
|
t2 = _make_track(2, "2026-03-14T02:00:00+00:00", title="Second")
|
|
t3 = _make_track(3, "2026-03-14T03:00:00+00:00", title="Third")
|
|
db.upsert_track(t1)
|
|
db.upsert_track(t2)
|
|
db.upsert_track(t3)
|
|
db.set_show_tracks(show.id, [t1.id, t2.id, t3.id])
|
|
db.set_show_tracks(show.id, [t1.id, t3.id])
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert len(tracks) == 2
|
|
assert tracks[0]["track_id"] == 1
|
|
assert tracks[0]["position"] == 1
|
|
assert tracks[1]["track_id"] == 3
|
|
assert tracks[1]["position"] == 2
|
|
|
|
|
|
def test_get_show_track_by_position(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00", title="First")
|
|
db.upsert_track(t1)
|
|
db.set_show_tracks(show.id, [t1.id])
|
|
result = db.get_show_track_by_position(show.id, 1)
|
|
assert result is not None
|
|
assert result["title"] == "First"
|
|
result_missing = db.get_show_track_by_position(show.id, 99)
|
|
assert result_missing is None
|
|
|
|
|
|
def test_list_shows(db):
|
|
_ = db.get_or_create_show(
|
|
datetime(2026, 3, 6, 3, 0, 0, tzinfo=timezone.utc),
|
|
datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
s2 = db.get_or_create_show(
|
|
datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc),
|
|
datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
shows = db.list_shows(limit=10, offset=0)
|
|
assert len(shows) == 2
|
|
assert shows[0].id == s2.id
|
|
|
|
|
|
def test_max_position_for_show(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
assert db.get_max_position(show.id) == 0
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00")
|
|
db.upsert_track(t1)
|
|
db.set_show_tracks(show.id, [t1.id])
|
|
assert db.get_max_position(show.id) == 1
|
|
|
|
|
|
def test_remove_show_track(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00")
|
|
t2 = _make_track(2, "2026-03-14T02:00:00+00:00")
|
|
t3 = _make_track(3, "2026-03-14T03:00:00+00:00")
|
|
db.upsert_track(t1)
|
|
db.upsert_track(t2)
|
|
db.upsert_track(t3)
|
|
db.set_show_tracks(show.id, [t1.id, t2.id, t3.id])
|
|
db.remove_show_track(show.id, 2)
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert len(tracks) == 2
|
|
assert tracks[0]["position"] == 1
|
|
assert tracks[0]["track_id"] == 1
|
|
assert tracks[1]["position"] == 2
|
|
assert tracks[1]["track_id"] == 3
|
|
|
|
|
|
def test_move_show_track(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00")
|
|
t2 = _make_track(2, "2026-03-14T02:00:00+00:00")
|
|
t3 = _make_track(3, "2026-03-14T03:00:00+00:00")
|
|
db.upsert_track(t1)
|
|
db.upsert_track(t2)
|
|
db.upsert_track(t3)
|
|
db.set_show_tracks(show.id, [t1.id, t2.id, t3.id])
|
|
db.move_show_track(show.id, track_id=3, new_position=1)
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert tracks[0]["track_id"] == 3
|
|
assert tracks[0]["position"] == 1
|
|
assert tracks[1]["track_id"] == 1
|
|
assert tracks[1]["position"] == 2
|
|
assert tracks[2]["track_id"] == 2
|
|
assert tracks[2]["position"] == 3
|
|
|
|
|
|
def test_add_track_to_show_at_position(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00")
|
|
t2 = _make_track(2, "2026-03-14T02:00:00+00:00")
|
|
t3 = _make_track(3, "2026-03-14T03:00:00+00:00")
|
|
db.upsert_track(t1)
|
|
db.upsert_track(t2)
|
|
db.upsert_track(t3)
|
|
db.set_show_tracks(show.id, [t1.id, t2.id])
|
|
db.add_track_to_show(show.id, track_id=3, position=2)
|
|
tracks = db.get_show_tracks(show.id)
|
|
assert len(tracks) == 3
|
|
assert tracks[0]["track_id"] == 1
|
|
assert tracks[1]["track_id"] == 3
|
|
assert tracks[2]["track_id"] == 2
|
|
|
|
|
|
def test_get_or_create_show_with_episode_number(db):
|
|
week_start = datetime(2026, 1, 8, 3, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 1, 15, 3, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end, episode_number=521)
|
|
assert show.episode_number == 521
|
|
show2 = db.get_or_create_show(week_start, week_end)
|
|
assert show2.id == show.id
|
|
assert show2.episode_number == 521
|
|
|
|
|
|
def test_get_or_create_show_updates_episode_number(db):
|
|
week_start = datetime(2026, 1, 8, 3, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 1, 15, 3, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
assert show.episode_number is None
|
|
show2 = db.get_or_create_show(week_start, week_end, episode_number=521)
|
|
assert show2.id == show.id
|
|
assert show2.episode_number == 521
|
|
|
|
|
|
def test_get_latest_episode_number(db):
|
|
assert db.get_latest_episode_number() is None
|
|
db.get_or_create_show(
|
|
datetime(2026, 1, 8, 3, 0, 0, tzinfo=timezone.utc),
|
|
datetime(2026, 1, 15, 3, 0, 0, tzinfo=timezone.utc),
|
|
episode_number=521,
|
|
)
|
|
assert db.get_latest_episode_number() == 521
|
|
db.get_or_create_show(
|
|
datetime(2026, 1, 15, 3, 0, 0, tzinfo=timezone.utc),
|
|
datetime(2026, 1, 22, 3, 0, 0, tzinfo=timezone.utc),
|
|
episode_number=522,
|
|
)
|
|
assert db.get_latest_episode_number() == 522
|
|
|
|
|
|
def test_update_show_episode_number(db):
|
|
week_start = datetime(2026, 1, 8, 3, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 1, 15, 3, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
assert show.episode_number is None
|
|
db.update_show_episode_number(show.id, 521)
|
|
show2 = db.get_or_create_show(week_start, week_end)
|
|
assert show2.episode_number == 521
|
|
|
|
|
|
def test_get_show_by_episode_number(db):
|
|
week_start = datetime(2026, 1, 8, 3, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 1, 15, 3, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end, episode_number=521)
|
|
result = db.get_show_by_episode_number(521)
|
|
assert result is not None
|
|
assert result.id == show.id
|
|
assert result.episode_number == 521
|
|
|
|
|
|
def test_get_show_by_episode_number_missing(db):
|
|
assert db.get_show_by_episode_number(999) is None
|
|
|
|
|
|
def test_has_track_in_show(db):
|
|
week_start = datetime(2026, 3, 13, 2, 0, 0, tzinfo=timezone.utc)
|
|
week_end = datetime(2026, 3, 20, 2, 0, 0, tzinfo=timezone.utc)
|
|
show = db.get_or_create_show(week_start, week_end)
|
|
t1 = _make_track(1, "2026-03-14T01:00:00+00:00")
|
|
db.upsert_track(t1)
|
|
db.set_show_tracks(show.id, [t1.id])
|
|
assert db.has_track_in_show(show.id, 1) is True
|
|
assert db.has_track_in_show(show.id, 999) is False
|