"""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, FrameTaskList, PhotoWidgetConfig, ServerSettings, TaskWidgetConfig, 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")} task_widget_columns = {c["name"] for c in inspector.get_columns("task_widget_configs")} 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 assert "name" in task_widget_columns # migration 19 assert "static_widget_configs" in inspector.get_table_names() # migration 20 assert "text_widget_configs" in inspector.get_table_names() # migration 21 text_widget_columns = {c["name"] for c in inspector.get_columns("text_widget_configs")} assert "font_family" in text_widget_columns # migration 22 assert "weather_widget_configs" in inspector.get_table_names() # migration 24 weather_widget_columns = {c["name"] for c in inspector.get_columns("weather_widget_configs")} assert {"mode", "provider", "city_latitude", "cities"} <= weather_widget_columns assert "battery_widget_configs" in inspector.get_table_names() # migration 25 battery_widget_columns = {c["name"] for c in inspector.get_columns("battery_widget_configs")} assert "mode" in battery_widget_columns widget_columns = {c["name"] for c in inspector.get_columns("widgets")} assert {"border_style", "border_thickness", "border_color_index"} <= widget_columns # migration 26 photo_widget_columns = {c["name"] for c in inspector.get_columns("photo_widget_configs")} assert "locked" in photo_widget_columns # migration 27 button_action_indexes = {idx["name"] for idx in inspector.get_indexes("frame_button_actions")} assert "ix_frame_button_actions_widget_button" in button_action_indexes # migration 28 assert {"hold_duration_ms", "next_hold_action", "back_hold_action", "last_cycled_layout_id"} <= frame_columns assert {"last_displayed_image", "last_displayed_at"} <= frame_columns # migration 30 assert "render_style" in weather_widget_columns # migration 31 # --- 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: # static_widget_configs/text_widget_configs are migration 20/21 # tables, saved_layouts/saved_layout_widgets/saved_layout_ # sources/saved_layout_button_actions are migration 23's, # weather_widget_configs is migration 24's, and # battery_widget_configs is migration 25's (all post-16, like the # rest of this list) -- dropped here too so a real version-15 # database is what's actually being simulated, not "version 15 # plus tables that wouldn't exist yet". Harmless to omit as long # as no migration after the one that creates a table also ALTERs # or re-CREATEs it (that's what let a create_all-based migration # go unlisted safely so far), but static_widget_configs/ # text_widget_configs/weather_widget_configs/battery_widget_ # configs all use a raw CREATE TABLE (not create_all -- see # migration 20's own docstring on why), so an already-present one # is a real "table already exists" collision, not a silent no-op. for table in ("frame_button_actions", "whiteboard_widget_configs", "task_widget_configs", "frame_task_lists", "calendar_widget_configs", "photo_widget_configs", "static_widget_configs", "text_widget_configs", "saved_layout_button_actions", "saved_layout_sources", "saved_layout_widgets", "saved_layouts", "weather_widget_configs", "battery_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_migration_28_dedupes_and_enforces_one_action_per_widget_per_button(db_session): """Exercises _migration_28's real ALTER path: a pre-existing database with more than one FrameButtonAction row bound to the same (widget, button) -- the old frame-level "Button assignments" card allowed this -- gets de-duped down to one row (MIN(id) survives) before the new unique index is created, rather than the migration failing outright on "UNIQUE constraint failed".""" frame = db_session.get(Frame, 1) widget = db_session.scalars(select(Widget).where(Widget.frame_id == frame.id)).first() with db_module.engine.begin() as conn: # The fresh-install create_all() copy already has the unique # index (it reflects models.py's current shape) -- drop it first # so the duplicate insert below doesn't just fail immediately, # simulating a real pre-migration-28 database. conn.execute(text("DROP INDEX IF EXISTS ix_frame_button_actions_widget_button")) conn.execute(text( "INSERT INTO frame_button_actions (frame_id, button, widget_id, action, sort_order, created_at) " "VALUES (:frame_id, 'next', :widget_id, 'back', 1, 0)" ), {"frame_id": frame.id, "widget_id": widget.id}) conn.execute(text("UPDATE schema_version SET version = 27")) run_migrations() with db_module.engine.connect() as conn: version = conn.execute(text("SELECT version FROM schema_version")).scalar() assert version == MIGRATIONS[-1][0] actions = db_session.scalars( select(FrameButtonAction).where(FrameButtonAction.widget_id == widget.id, FrameButtonAction.button == "next") ).all() assert len(actions) == 1 inspector = inspect(db_module.engine) indexes = {idx["name"] for idx in inspector.get_indexes("frame_button_actions")} assert "ix_frame_button_actions_widget_button" in indexes def test_migration_29_adds_hold_action_columns_to_an_existing_database(db_session): """Exercises _migration_29's real guarded ALTER path (frames isn't dropped/recreated by the pre-widget-system replay tests, so its columns must be added defensively, same reasoning as migration 26/27's own comments).""" with db_module.engine.begin() as conn: conn.execute(text("UPDATE schema_version SET version = 28")) run_migrations() with db_module.engine.connect() as conn: version = conn.execute(text("SELECT version FROM schema_version")).scalar() assert version == MIGRATIONS[-1][0] frame = db_session.get(Frame, 1) assert frame.hold_duration_ms == 3000 assert frame.next_hold_action is None assert frame.back_hold_action is None assert frame.last_cycled_layout_id is None def test_migration_30_adds_now_displaying_columns_to_an_existing_database(db_session): """Exercises _migration_30's real guarded ALTER path (frames isn't dropped/recreated by the pre-widget-system replay tests, so its columns must be added defensively, same reasoning as migration 26/27/29's own comments).""" with db_module.engine.begin() as conn: conn.execute(text("UPDATE schema_version SET version = 29")) run_migrations() with db_module.engine.connect() as conn: version = conn.execute(text("SELECT version FROM schema_version")).scalar() assert version == MIGRATIONS[-1][0] frame = db_session.get(Frame, 1) assert frame.last_displayed_image is None assert frame.last_displayed_at == 0.0 def test_migration_17_and_18_extract_tasks_into_a_standalone_multi_list_widget(db_session): """Exercises _migration_17 and _migration_18's actual data-extraction SQL back to back (the real "existing widget-system database upgrading past both migrations" scenario, since both apply in the same run_migrations() call here): a calendar_widget_configs row in its pre-17 shape (tasks_* columns still present, still holding a configured task source) should come out the other side as a sibling `tasks` widget with that source carried forward as its first FrameTaskList row (migration 17's extraction, then migration 18's further extraction of the single-source TaskWidgetConfig it produces into FrameTaskList), with calendar_widget_configs no longer having tasks_* columns and TaskWidgetConfig no longer having user_id/ calendar_key columns either.""" with db_module.engine.begin() as conn: # static_widget_configs/text_widget_configs/saved_layout_*/ # weather_widget_configs/battery_widget_configs dropped too -- # see the comment on the identical setup in # test_migration_16_raw_sql_path_applies_to_an_existing_pre_widget_database above (migrations # 20/21/23/24/25's raw CREATE TABLE collides with an already-present table # otherwise, since this test replays 17 through 25 and none of # these tables would really exist yet at a genuine pre-migration-17 # schema_version). conn.execute(text("DROP TABLE task_widget_configs")) conn.execute(text("DROP TABLE frame_task_lists")) conn.execute(text("DROP TABLE calendar_widget_configs")) conn.execute(text("DROP TABLE static_widget_configs")) conn.execute(text("DROP TABLE text_widget_configs")) conn.execute(text("DROP TABLE saved_layout_button_actions")) conn.execute(text("DROP TABLE saved_layout_sources")) conn.execute(text("DROP TABLE saved_layout_widgets")) conn.execute(text("DROP TABLE saved_layouts")) conn.execute(text("DROP TABLE weather_widget_configs")) conn.execute(text("DROP TABLE battery_widget_configs")) conn.execute(text( "CREATE TABLE calendar_widget_configs (" "widget_id INTEGER PRIMARY KEY REFERENCES widgets(id) ON DELETE CASCADE, " "view TEXT NOT NULL DEFAULT 'agenda', " "week_start INTEGER NOT NULL DEFAULT 0, " "browse_offset INTEGER NOT NULL DEFAULT 0, " "checked_at REAL NOT NULL DEFAULT 0.0, " "cached_events TEXT, " "fetch_summary TEXT NOT NULL DEFAULT '', " "weather_enabled INTEGER NOT NULL DEFAULT 0, " "weather_units TEXT NOT NULL DEFAULT 'fahrenheit', " "weather_cities TEXT, " "weather_checked_at REAL NOT NULL DEFAULT 0.0, " "weather_cached TEXT, " "week_days INTEGER NOT NULL DEFAULT 7, " "week_layout TEXT NOT NULL DEFAULT 'horizontal', " "week_start_offset INTEGER NOT NULL DEFAULT 0, " "tasks_enabled INTEGER NOT NULL DEFAULT 0, " "tasks_user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, " "tasks_calendar_key TEXT, " "tasks_checked_at REAL NOT NULL DEFAULT 0.0, " "tasks_cached TEXT)" )) conn.execute(text("UPDATE schema_version SET version = 16")) user = make_user(db_session, "task-owner") user_id = user.id # Frame #1's auto-migrated widget is a full-panel "photos" one -- # remove it to free up grid space for the calendar widget below (and # for the tasks widget this migration is expected to carve out). db_session.query(Widget).filter_by(frame_id=1, widget_type="photos").delete() db_session.commit() calendar_widget = Widget(frame_id=1, widget_type="calendar", x=0, y=0, w=3, h=2, sort_order=1, created_at=time.time()) db_session.add(calendar_widget) db_session.flush() calendar_widget_id = calendar_widget.id db_session.commit() with db_module.engine.begin() as conn: conn.execute(text( "INSERT INTO calendar_widget_configs " "(widget_id, tasks_enabled, tasks_user_id, tasks_calendar_key, tasks_checked_at, tasks_cached) " "VALUES (:widget_id, 1, :user_id, 'caldav:/some/tasks/', 123.0, '[{\"summary\": \"Buy milk\"}]')" ), {"widget_id": calendar_widget_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] columns = {c["name"] for c in inspect(db_module.engine).get_columns("calendar_widget_configs")} assert not any(c.startswith("tasks_") for c in columns) task_cfg_columns = {c["name"] for c in inspect(db_module.engine).get_columns("task_widget_configs")} assert "user_id" not in task_cfg_columns and "calendar_key" not in task_cfg_columns assert "show_completed" in task_cfg_columns task_widgets = db_session.scalars( select(Widget).where(Widget.frame_id == 1, Widget.widget_type == "tasks") ).all() assert len(task_widgets) == 1 task_widget = task_widgets[0] cfg = db_session.get(TaskWidgetConfig, task_widget.id) assert cfg.checked_at == 123.0 assert cfg.cached == [{"summary": "Buy milk"}] assert cfg.show_completed is False task_list = db_session.scalars( select(FrameTaskList).where(FrameTaskList.widget_id == task_widget.id) ).one() assert task_list.user_id == user_id assert task_list.calendar_key == "caldav:/some/tasks/" assert task_list.included is True # Auto-placed without overlapping the calendar widget it was split from. cal = db_session.get(Widget, calendar_widget_id) assert not grid.overlaps((cal.x, cal.y, cal.w, cal.h), (task_widget.x, task_widget.y, task_widget.w, task_widget.h)) 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: # static_widget_configs/text_widget_configs/saved_layout_*/ # weather_widget_configs dropped too -- see the comment on the # identical setup in # test_migration_16_raw_sql_path_applies_to_an_existing_pre_widget_database above. for table in ("frame_button_actions", "whiteboard_widget_configs", "task_widget_configs", "frame_task_lists", "calendar_widget_configs", "photo_widget_configs", "static_widget_configs", "text_widget_configs", "saved_layout_button_actions", "saved_layout_sources", "saved_layout_widgets", "saved_layouts", "weather_widget_configs", "battery_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