espresso_frame/server/tests/test_global_actions.py

"""Hold-for-global-action (see app/global_actions.py): the
Configuration-tab save path (hold_duration_ms/next_hold_action/
back_hold_action, folded into api_frames.py's api_config_save), the
device-facing dispatch (/frame/global-next, /frame/global-back in
routers/device.py), and the three registry actions themselves
(cycle_layout, refresh_all_widgets, toggle_all_photo_locks)."""

from __future__ import annotations

import time

from app import global_actions, widgets
from app.models import (
    CalendarWidgetConfig,
    Frame,
    PhotoWidgetConfig,
    SavedLayout,
    Widget,
    WhiteboardWidgetConfig,
)

from .conftest import claim_device, csrf_headers, link_user, login, make_user

EXPECTED_BYTES = 800 * 480 // 2


# --- Configuration-tab save (api_frames.py's api_config_save) --------------

def test_save_hold_duration_clamps_to_range(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})

    resp = client.post("/api/frames/1/config", data={"hold_duration_ms": "999"}, headers=csrf_headers(client))
    assert resp.status_code == 200, resp.text
    assert db_session.get(Frame, 1).hold_duration_ms == 3000  # MIN_HOLD_DURATION_MS

    resp = client.post("/api/frames/1/config", data={"hold_duration_ms": "999999"}, headers=csrf_headers(client))
    assert resp.status_code == 200, resp.text
    assert db_session.get(Frame, 1).hold_duration_ms == 10000  # MAX_HOLD_DURATION_MS


def test_save_sets_and_clears_hold_actions(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})

    resp = client.post(
        "/api/frames/1/config",
        data={"next_hold_action": "cycle_layout", "back_hold_action": "refresh_all_widgets"},
        headers=csrf_headers(client),
    )
    assert resp.status_code == 200, resp.text
    frame = db_session.get(Frame, 1)
    assert frame.next_hold_action == "cycle_layout"
    assert frame.back_hold_action == "refresh_all_widgets"

    resp = client.post(
        "/api/frames/1/config", data={"next_hold_action": "", "back_hold_action": ""},
        headers=csrf_headers(client),
    )
    assert resp.status_code == 200, resp.text
    db_session.refresh(frame)
    assert frame.next_hold_action is None
    assert frame.back_hold_action is None


def test_save_unknown_action_clears_to_none(client, db_session):
    """Not a 400 -- same silent-normalize posture as this endpoint's
    other enum-ish fields (orientation, timezone)."""
    client.post("/setup", data={"username": "alice", "password": "hunter22"})

    resp = client.post(
        "/api/frames/1/config", data={"next_hold_action": "not_a_real_action"},
        headers=csrf_headers(client),
    )
    assert resp.status_code == 200, resp.text
    assert db_session.get(Frame, 1).next_hold_action is None


def test_save_linked_but_not_controlling_user_409s(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    bob = make_user(db_session, "bob")
    frame = db_session.get(Frame, 1)
    link_user(db_session, bob, frame)

    client.cookies.clear()
    login(client, "bob")
    resp = client.post(
        "/api/frames/1/config", data={"next_hold_action": "cycle_layout"}, headers=csrf_headers(client)
    )
    assert resp.status_code == 409
    assert resp.json()["detail"]["error"] == "not_controller"


def test_save_unrelated_user_404s(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    make_user(db_session, "mallory")

    client.cookies.clear()
    login(client, "mallory")
    resp = client.post(
        "/api/frames/1/config", data={"next_hold_action": "cycle_layout"}, headers=csrf_headers(client)
    )
    assert resp.status_code == 404
    assert db_session.get(Frame, 1).next_hold_action is None


def test_save_logged_out_401s(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    client.cookies.clear()
    resp = client.post("/api/frames/1/config", data={"next_hold_action": "cycle_layout"})
    assert resp.status_code == 401


# --- Device dispatch (/frame/global-next, /frame/global-back) --------------

def test_global_next_is_a_noop_when_unset(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    creds = claim_device(db_session, db_session.get(Frame, 1))
    resp = client.post(f"/frame/global-next?{creds}")
    assert resp.status_code == 200
    assert len(resp.content) == EXPECTED_BYTES


def test_global_next_runs_the_configured_action(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)
    photo_widget_id = db_session.query(Widget).filter_by(frame_id=frame.id, widget_type="photos").one().id
    frame.next_hold_action = "toggle_all_photo_locks"
    creds = claim_device(db_session, frame)

    resp = client.post(f"/frame/global-next?{creds}")
    assert resp.status_code == 200
    assert len(resp.content) == EXPECTED_BYTES
    assert db_session.get(PhotoWidgetConfig, photo_widget_id).locked is True


def test_global_back_runs_the_configured_action(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)
    photo_widget_id = db_session.query(Widget).filter_by(frame_id=frame.id, widget_type="photos").one().id
    frame.back_hold_action = "toggle_all_photo_locks"
    creds = claim_device(db_session, frame)

    resp = client.post(f"/frame/global-back?{creds}")
    assert resp.status_code == 200
    assert db_session.get(PhotoWidgetConfig, photo_widget_id).locked is True


def test_global_next_with_an_unrecognized_stored_action_is_a_noop(client, db_session):
    """Defensive -- a value that stopped being a valid registry key
    (e.g. after a downgrade) shouldn't 500 the whole request."""
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)
    frame.next_hold_action = "no_longer_exists"
    creds = claim_device(db_session, frame)

    resp = client.post(f"/frame/global-next?{creds}")
    assert resp.status_code == 200
    assert len(resp.content) == EXPECTED_BYTES


# --- Registry: toggle_all_photo_locks ---------------------------------------

def test_toggle_all_photo_locks_locks_then_unlocks(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)
    photo_widget_id = db_session.query(Widget).filter_by(frame_id=frame.id, widget_type="photos").one().id

    global_actions.toggle_all_photo_locks(db_session, frame)
    assert db_session.get(PhotoWidgetConfig, photo_widget_id).locked is True

    global_actions.toggle_all_photo_locks(db_session, frame)
    assert db_session.get(PhotoWidgetConfig, photo_widget_id).locked is False


def test_toggle_all_photo_locks_noop_with_no_photo_widgets(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)
    widget = db_session.query(Widget).filter_by(frame_id=frame.id, widget_type="photos").one()
    db_session.delete(widget)
    db_session.commit()

    global_actions.toggle_all_photo_locks(db_session, frame)  # should not raise


# --- Registry: refresh_all_widgets ------------------------------------------

def test_refresh_all_widgets_only_calls_check_now_capable_types(client, db_session, monkeypatch):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)
    photo_widget = db_session.query(Widget).filter_by(frame_id=frame.id, widget_type="photos").one()

    board_widget = Widget(frame_id=frame.id, widget_type="whiteboard", x=0, y=0, w=1, h=1,
                          sort_order=1, created_at=time.time())
    db_session.add(board_widget)
    db_session.flush()
    db_session.add(WhiteboardWidgetConfig(widget_id=board_widget.id))
    db_session.commit()

    calls = []
    monkeypatch.setattr(
        widgets.whiteboard, "get_or_refresh_whiteboard_for_widget",
        lambda db, frame, widget, force=False: calls.append((widget.id, force)),
    )

    global_actions.refresh_all_widgets(db_session, frame)

    # Only the whiteboard widget has a check_now action -- the photos
    # widget (no check_now in its ACTIONS) is left untouched.
    assert calls == [(board_widget.id, True)]
    assert photo_widget.id not in [c[0] for c in calls]


def test_refresh_all_widgets_one_failure_does_not_block_others(client, db_session, monkeypatch):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)

    board_a = Widget(frame_id=frame.id, widget_type="whiteboard", x=0, y=0, w=1, h=1,
                     sort_order=1, created_at=time.time())
    board_b = Widget(frame_id=frame.id, widget_type="whiteboard", x=1, y=0, w=1, h=1,
                     sort_order=2, created_at=time.time())
    db_session.add_all([board_a, board_b])
    db_session.flush()
    db_session.add_all([WhiteboardWidgetConfig(widget_id=board_a.id), WhiteboardWidgetConfig(widget_id=board_b.id)])
    db_session.commit()

    calls = []

    def _flaky(db, frame, widget, force=False):
        if widget.id == board_a.id:
            raise RuntimeError("simulated fetch failure")
        calls.append(widget.id)

    monkeypatch.setattr(widgets.whiteboard, "get_or_refresh_whiteboard_for_widget", _flaky)

    global_actions.refresh_all_widgets(db_session, frame)  # should not raise
    assert calls == [board_b.id]


# --- Registry: cycle_layout --------------------------------------------------

def test_cycle_layout_noop_when_frame_unclaimed(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)
    frame.owner_user_id = None
    db_session.commit()

    global_actions.cycle_layout(db_session, frame)  # should not raise
    assert db_session.query(Widget).filter_by(frame_id=frame.id).count() == 1  # untouched


def test_cycle_layout_noop_when_no_compatible_layouts(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)

    global_actions.cycle_layout(db_session, frame)  # no saved layouts at all yet
    assert db_session.query(Widget).filter_by(frame_id=frame.id).count() == 1


def test_cycle_layout_applies_and_wraps_around(client, db_session):
    client.post("/setup", data={"username": "alice", "password": "hunter22"})
    frame = db_session.get(Frame, 1)

    resp_a = client.post("/api/frames/1/layouts", json={"name": "Layout A"}, headers=csrf_headers(client))
    assert resp_a.status_code == 200, resp_a.text
    layout_a_id = resp_a.json()["id"]

    # A second, differently-shaped widget arrangement to save as "Layout B".
    photo_widget = db_session.query(Widget).filter_by(frame_id=frame.id, widget_type="photos").one()
    photo_widget.w = 4
    cal_widget = Widget(frame_id=frame.id, widget_type="calendar", x=4, y=0, w=4, h=5,
                        sort_order=1, created_at=time.time())
    db_session.add(cal_widget)
    db_session.flush()
    db_session.add(CalendarWidgetConfig(widget_id=cal_widget.id))
    db_session.commit()
    resp_b = client.post("/api/frames/1/layouts", json={"name": "Layout B"}, headers=csrf_headers(client))
    assert resp_b.status_code == 200, resp_b.text
    layout_b_id = resp_b.json()["id"]

    # A layout with the wrong grid dims must never be selected.
    incompatible = SavedLayout(user_id=frame.owner_user_id, name="Wrong size", cols=1, rows=1,
                               created_at=time.time(), updated_at=time.time())
    db_session.add(incompatible)
    db_session.commit()

    global_actions.cycle_layout(db_session, frame)
    db_session.refresh(frame)
    assert frame.last_cycled_layout_id == layout_a_id

    global_actions.cycle_layout(db_session, frame)
    db_session.refresh(frame)
    assert frame.last_cycled_layout_id == layout_b_id

    global_actions.cycle_layout(db_session, frame)  # wraps back to the first
    db_session.refresh(frame)
    assert frame.last_cycled_layout_id == layout_a_id