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,
    PhotoWidgetConfig,
    ServerSettings,
    Widget,
    WhiteboardWidgetConfig,
)


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."""
    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("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.commit()
    frame_id = frame.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"]