"""migration.py structural sanity: MIGRATIONS is well-formed, a fresh install lands on the latest schema version with frame #1 + server settings seeded, and re-running run_migrations() is a true no-op (it's called unconditionally at every app.main import -- see main.py -- so it has to tolerate being invoked against an already-current database).""" from __future__ import annotations import time from sqlalchemy import inspect, select, text from app import db as db_module from app import grid from app.migration import MIGRATIONS, run_migrations from app.models import ( CalendarWidgetConfig, Frame, FrameButtonAction, FrameCalendar, PhotoWidgetConfig, ServerSettings, Widget, WhiteboardWidgetConfig, ) from .conftest import make_user def test_migrations_list_is_sequential_and_unique(): versions = [v for v, _ in MIGRATIONS] assert versions == sorted(versions) assert len(versions) == len(set(versions)) assert versions == list(range(1, len(versions) + 1)) def test_fresh_install_lands_on_latest_version(db_session): with db_module.engine.connect() as conn: row = conn.execute(text("SELECT version FROM schema_version")).fetchone() assert row is not None assert row[0] == MIGRATIONS[-1][0] def test_fresh_install_seeds_frame_one_and_server_settings(db_session): frame = db_session.get(Frame, 1) assert frame is not None assert frame.name settings = db_session.get(ServerSettings, 1) assert settings is not None def test_rerunning_migrations_is_a_no_op(db_session): frame_count_before = len(db_session.query(Frame).all()) run_migrations() run_migrations() frame_count_after = len(db_session.query(Frame).all()) assert frame_count_before == frame_count_after == 1 def test_expected_columns_exist_on_current_schema(): """A light spot-check, not exhaustive -- one column from a handful of the more recent migrations, to catch an ALTER that silently didn't apply (e.g. a typo'd table/column name in a migration function).""" inspector = inspect(db_module.engine) user_columns = {c["name"] for c in inspector.get_columns("users")} frame_columns = {c["name"] for c in inspector.get_columns("frames")} assert "webdav_base_url" in user_columns # migration 15 assert "webdav_username" in user_columns # migration 14 assert "calendar_caldav_url" in user_columns assert "whiteboard_cached_image" in frame_columns # migration 14 assert "calendar_week_start_offset" in frame_columns # --- widget system backfill (migration 16 + _ensure_widgets_backfilled) --- def test_fresh_install_backfills_one_photos_widget_with_default_buttons(db_session): frame = db_session.get(Frame, 1) assert frame.mode == "photos" widgets = db_session.scalars(select(Widget).where(Widget.frame_id == frame.id)).all() assert len(widgets) == 1 widget = widgets[0] assert widget.widget_type == "photos" assert (widget.x, widget.y, widget.w, widget.h) == grid.full_panel_rect(frame.orientation) config = db_session.get(PhotoWidgetConfig, widget.id) assert config is not None assert config.album_id == frame.album_id actions = db_session.scalars(select(FrameButtonAction).where(FrameButtonAction.frame_id == frame.id)).all() assert {a.button: a.action for a in actions} == {"next": "advance", "back": "back"} assert all(a.widget_id == widget.id for a in actions) def test_rerunning_migrations_does_not_duplicate_widgets(db_session): run_migrations() run_migrations() widgets = db_session.scalars(select(Widget).where(Widget.frame_id == 1)).all() assert len(widgets) == 1 def test_calendar_photo_inlay_frame_backfills_into_two_widgets(db_session): """Reproduces the old fixed 50/50 inlay split as two independent, non-overlapping widgets instead of silently dropping the photo half on upgrade -- see models.py's CalendarWidgetConfig docstring.""" frame = Frame( name="Inlay Frame", device_token="tok-inlay", manage_token="mtok-inlay", mode="calendar", orientation="landscape", calendar_view="week", calendar_photo_inlay=True, album_id="album-123", current_asset_id="asset-1", queue=["asset-1", "asset-2"], created_at=time.time(), ) db_session.add(frame) db_session.commit() run_migrations() widgets = db_session.scalars( select(Widget).where(Widget.frame_id == frame.id).order_by(Widget.sort_order) ).all() assert len(widgets) == 2 cal_widget, photo_widget = widgets assert cal_widget.widget_type == "calendar" assert photo_widget.widget_type == "photos" cal_rect = (cal_widget.x, cal_widget.y, cal_widget.w, cal_widget.h) photo_rect = (photo_widget.x, photo_widget.y, photo_widget.w, photo_widget.h) assert not grid.overlaps(cal_rect, photo_rect) assert cal_widget.w + photo_widget.w == grid.grid_dims("landscape")[0] assert cal_widget.h == photo_widget.h == grid.grid_dims("landscape")[1] cal_config = db_session.get(CalendarWidgetConfig, cal_widget.id) assert cal_config.view == "week" photo_config = db_session.get(PhotoWidgetConfig, photo_widget.id) assert photo_config.album_id == "album-123" assert photo_config.current_asset_id == "asset-1" assert photo_config.queue == ["asset-1", "asset-2"] actions = db_session.scalars(select(FrameButtonAction).where(FrameButtonAction.frame_id == frame.id)).all() assert {a.button: a.action for a in actions} == {"next": "advance", "back": "back"} assert all(a.widget_id == cal_widget.id for a in actions) def test_whiteboard_frame_backfills_check_now_on_both_buttons(db_session): frame = Frame( name="WB Frame", device_token="tok-wb", manage_token="mtok-wb", mode="whiteboard", orientation="portrait", whiteboard_url="https://example.com/board.whiteboard", created_at=time.time(), ) db_session.add(frame) db_session.commit() run_migrations() widgets = db_session.scalars(select(Widget).where(Widget.frame_id == frame.id)).all() assert len(widgets) == 1 widget = widgets[0] assert widget.widget_type == "whiteboard" assert (widget.x, widget.y, widget.w, widget.h) == grid.full_panel_rect("portrait") config = db_session.get(WhiteboardWidgetConfig, widget.id) assert config.url == "https://example.com/board.whiteboard" actions = db_session.scalars(select(FrameButtonAction).where(FrameButtonAction.frame_id == frame.id)).all() assert {a.action for a in actions} == {"check_now"} def test_migration_16_raw_sql_path_applies_to_an_existing_pre_widget_database(db_session): """Exercises _migration_16's actual CREATE TABLE statements (the real "existing production database upgrading past this migration" scenario) rather than the fresh-install create_all() shortcut, which every other test in this file goes through instead. frame_calendars also gets rebuilt back to its pre-rekey (frame_id-keyed) shape, matching what _migration_9/_migration_11 originally produced, so _ensure_frame_calendars_rekeyed has a real frame_id-shaped table to migrate.""" with db_module.engine.begin() as conn: for table in ("frame_button_actions", "whiteboard_widget_configs", "calendar_widget_configs", "photo_widget_configs", "widgets"): conn.execute(text(f"DROP TABLE {table}")) conn.execute(text("DROP TABLE frame_calendars")) conn.execute(text( "CREATE TABLE frame_calendars (" "id INTEGER PRIMARY KEY, " "frame_id INTEGER NOT NULL REFERENCES frames(id) ON DELETE CASCADE, " "user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, " "calendar_key TEXT NOT NULL, " "calendar_label TEXT NOT NULL DEFAULT '', " "included INTEGER NOT NULL DEFAULT 1, " "color_index INTEGER)" )) conn.execute(text( "CREATE UNIQUE INDEX ix_frame_calendars_unique ON frame_calendars (frame_id, user_id, calendar_key)" )) conn.execute(text("UPDATE schema_version SET version = 15")) frame = Frame( name="Upgrading Frame", device_token="tok-up", manage_token="mtok-up", mode="photos", album_id="legacy-album", current_asset_id="legacy-asset", queue=["legacy-asset", "next-asset"], created_at=time.time(), ) db_session.add(frame) db_session.flush() user = make_user(db_session, "legacy-owner") db_session.commit() frame_id, user_id = frame.id, user.id # An orphaned frame_calendars row (this frame's mode was never # "calendar", so it has no calendar widget for _ensure_frame_ # calendars_rekeyed to attach it to) -- exercises that it's dropped # cleanly rather than raising. with db_module.engine.begin() as conn: conn.execute(text( "INSERT INTO frame_calendars (frame_id, user_id, calendar_key, calendar_label, included) " "VALUES (:frame_id, :user_id, 'ics', 'Legacy Cal', 1)" ), {"frame_id": frame_id, "user_id": user_id}) run_migrations() with db_module.engine.begin() as conn: version = conn.execute(text("SELECT version FROM schema_version")).scalar() assert version == MIGRATIONS[-1][0] widgets = db_session.scalars(select(Widget).where(Widget.frame_id == frame_id)).all() assert len(widgets) == 1 config = db_session.get(PhotoWidgetConfig, widgets[0].id) assert config.album_id == "legacy-album" assert config.current_asset_id == "legacy-asset" assert config.queue == ["legacy-asset", "next-asset"] def test_frame_calendars_rekey_attaches_existing_rows_to_their_calendar_widget(db_session): """A frame whose mode was "calendar" (not "photos") gets a calendar widget from _ensure_widgets_backfilled -- _ensure_frame_calendars_ rekeyed should then attach the pre-existing frame_id-keyed FrameCalendar row inserted below to that widget's id, not drop it. This is the regression case for the ordering bug the two migration tests above's setup was designed to catch: the rekey must run AFTER the widget backfill, not as a numbered migration racing ahead of it (see _ensure_frame_calendars_rekeyed's own docstring).""" with db_module.engine.begin() as conn: for table in ("frame_button_actions", "whiteboard_widget_configs", "calendar_widget_configs", "photo_widget_configs", "widgets"): conn.execute(text(f"DROP TABLE {table}")) conn.execute(text("DROP TABLE frame_calendars")) conn.execute(text( "CREATE TABLE frame_calendars (" "id INTEGER PRIMARY KEY, " "frame_id INTEGER NOT NULL REFERENCES frames(id) ON DELETE CASCADE, " "user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, " "calendar_key TEXT NOT NULL, " "calendar_label TEXT NOT NULL DEFAULT '', " "included INTEGER NOT NULL DEFAULT 1, " "color_index INTEGER)" )) conn.execute(text( "CREATE UNIQUE INDEX ix_frame_calendars_unique ON frame_calendars (frame_id, user_id, calendar_key)" )) conn.execute(text("UPDATE schema_version SET version = 15")) frame = Frame( name="Calendar Frame", device_token="tok-cal", manage_token="mtok-cal", mode="calendar", created_at=time.time(), ) db_session.add(frame) db_session.flush() user = make_user(db_session, "cal-owner") db_session.commit() frame_id, user_id = frame.id, user.id with db_module.engine.begin() as conn: conn.execute(text( "INSERT INTO frame_calendars (frame_id, user_id, calendar_key, calendar_label, included, color_index) " "VALUES (:frame_id, :user_id, 'ics', 'Legacy Cal', 1, 3)" ), {"frame_id": frame_id, "user_id": user_id}) run_migrations() widget = db_session.scalars( select(Widget).where(Widget.frame_id == frame_id, Widget.widget_type == "calendar") ).one() row = db_session.scalars(select(FrameCalendar).where(FrameCalendar.widget_id == widget.id)).one() assert row.user_id == user_id assert row.calendar_key == "ics" assert row.calendar_label == "Legacy Cal" assert row.included is True assert row.color_index == 3