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): s1 = 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_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