f2f4f8ccee
Full Refresh INSERTs a per-track year (from file tags) into tracks.year, but that column was only ever in the live INSERT — never in CREATE TABLE and never in a migration. So on EVERY db (old and current — verified the shipped music_library.db lacks it too) every Full Refresh track insert hard-failed with 'table tracks has no column named year', importing 0 tracks while artists/albums succeeded. Fix (additive + nullable, nothing reads it but the writer): - add year INTEGER to the tracks CREATE TABLE (new DBs) - ALTER it onto existing tracks tables in _ensure_core_media_schema_columns (the repair backstop that already runs every init), right beside the file_size repair Tests (tests/test_tracks_year_migration.py): fresh-DB has it, nullable, idempotent, ALTERs onto an old year-less table, and a regression that the exact Full Refresh insert fails before the repair and succeeds after.
104 lines
4 KiB
Python
104 lines
4 KiB
Python
"""Migration + regression test for the per-track ``year`` column (#910).
|
|
|
|
Full Refresh INSERTs a per-track ``year`` (read from file tags), but the column was
|
|
only ever present in that live INSERT — never in CREATE TABLE and never in a
|
|
migration. So on every DB (old *and* current) the Full Refresh track insert
|
|
hard-failed with ``table tracks has no column named year``, importing 0 tracks
|
|
while artists/albums imported fine.
|
|
|
|
The repair backstop (``_ensure_core_media_schema_columns``) must ALTER ``year``
|
|
onto any tracks table that lacks it. Additive + nullable; nothing reads it except
|
|
the writer, so backfilling it on every existing DB is safe.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from database.music_database import MusicDatabase
|
|
|
|
|
|
def _track_cols(cur):
|
|
cur.execute("PRAGMA table_info(tracks)")
|
|
return {c[1] for c in cur.fetchall()}
|
|
|
|
|
|
# An upgraded tracks table that has every Full Refresh insert column EXCEPT year
|
|
# (mirrors the #910 reporter exactly: "64 columns, only year absent"). Text ids
|
|
# match the live schema after the id->TEXT migration.
|
|
_OLD_TRACKS = (
|
|
"CREATE TABLE tracks (id TEXT PRIMARY KEY, album_id TEXT, artist_id TEXT, "
|
|
"title TEXT, track_number INTEGER, disc_number INTEGER, duration INTEGER, "
|
|
"file_path TEXT, bitrate INTEGER, server_source TEXT, "
|
|
"created_at TIMESTAMP, updated_at TIMESTAMP)"
|
|
)
|
|
|
|
# The exact Full Refresh insert from web_server.py (the statement that failed).
|
|
_FULL_REFRESH_INSERT = (
|
|
"INSERT OR IGNORE INTO tracks (id, album_id, artist_id, title, track_number, disc_number, "
|
|
"duration, file_path, bitrate, year, server_source, created_at, updated_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'soulsync', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
|
|
)
|
|
_ROW = ('t1::soulsync', 'a1', 'ar1', 'Song', 1, 1, 200000, '/music/song.flac', 1000, 2009)
|
|
|
|
|
|
def test_fresh_db_has_year_column(tmp_path):
|
|
db = MusicDatabase(str(tmp_path / "m.db"))
|
|
cur = db._get_connection().cursor()
|
|
assert 'year' in _track_cols(cur)
|
|
|
|
|
|
def test_year_column_is_nullable(tmp_path):
|
|
db = MusicDatabase(str(tmp_path / "m.db"))
|
|
cur = db._get_connection().cursor()
|
|
cur.execute("PRAGMA table_info(tracks)")
|
|
info = {c[1]: c for c in cur.fetchall()} # name -> (cid, name, type, notnull, dflt, pk)
|
|
assert info['year'][3] == 0 # nullable -> safe to add to a populated table
|
|
|
|
|
|
def test_migration_is_idempotent(tmp_path):
|
|
db = MusicDatabase(str(tmp_path / "m.db"))
|
|
cur = db._get_connection().cursor()
|
|
before = _track_cols(cur)
|
|
# Re-running must not raise (the PRAGMA guard skips the existing column).
|
|
db._ensure_core_media_schema_columns(cur)
|
|
db._ensure_core_media_schema_columns(cur)
|
|
assert _track_cols(cur) == before
|
|
assert 'year' in _track_cols(cur)
|
|
|
|
|
|
def test_migration_adds_year_to_old_tracks_table(tmp_path):
|
|
conn = sqlite3.connect(str(tmp_path / "old.db"))
|
|
conn.execute(_OLD_TRACKS)
|
|
conn.commit()
|
|
cur = conn.cursor()
|
|
assert 'year' not in _track_cols(cur)
|
|
|
|
db = MusicDatabase(str(tmp_path / "scratch.db"))
|
|
db._ensure_core_media_schema_columns(cur)
|
|
conn.commit()
|
|
|
|
assert 'year' in _track_cols(cur)
|
|
|
|
|
|
def test_full_refresh_insert_fails_before_repair_and_succeeds_after(tmp_path):
|
|
"""Regression for #910: the real Full Refresh track insert hard-fails on a
|
|
year-less tracks table, then succeeds once the repair has added the column."""
|
|
conn = sqlite3.connect(str(tmp_path / "old.db"))
|
|
conn.execute(_OLD_TRACKS)
|
|
conn.commit()
|
|
cur = conn.cursor()
|
|
|
|
# Before the fix: the live insert blows up exactly as the issue reports.
|
|
with pytest.raises(sqlite3.OperationalError, match="no column named year"):
|
|
cur.execute(_FULL_REFRESH_INSERT, _ROW)
|
|
|
|
# Apply the repair backstop, then the same insert must succeed and persist year.
|
|
db = MusicDatabase(str(tmp_path / "scratch.db"))
|
|
db._ensure_core_media_schema_columns(cur)
|
|
cur.execute(_FULL_REFRESH_INSERT, _ROW)
|
|
conn.commit()
|
|
cur.execute("SELECT year FROM tracks WHERE id = 't1::soulsync'")
|
|
assert cur.fetchone()[0] == 2009
|