espresso_frame/server/tests/test_migrations.py

"""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 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_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