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,
    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