"""routers/api_layouts.py -- named, user-owned saved layouts. Save/apply snapshot/restore a frame's whole widget arrangement (placement, per-type settings, calendar/task sources, button-action bindings); list/rename/ delete operate purely on "is this your own saved layout", independent of any frame -- see models.SavedLayout's docstring for why layouts are user-owned rather than frame-owned. Frame #1's auto-migrated widget (see migration.py's backfill) is a single full-panel (0, 0, 8, 5) photos widget -- same starting point test_widget_placement.py's tests use.""" from __future__ import annotations import time from app.models import ( CalendarWidgetConfig, Frame, FrameButtonAction, FrameCalendar, FrameTaskList, PhotoWidgetConfig, SavedLayout, SavedLayoutButtonAction, SavedLayoutSource, SavedLayoutWidget, TaskWidgetConfig, User, Widget, WeatherWidgetConfig, ) from .conftest import csrf_headers, link_user, login, make_user def _widget_id(db_session, widget_type="photos", frame_id=1) -> int: return db_session.query(Widget).filter_by(frame_id=frame_id, widget_type=widget_type).one().id def _add_calendar_widget(db_session, frame_id=1, x=0, y=0, w=3, h=2, sort_order=1) -> Widget: widget = Widget(frame_id=frame_id, widget_type="calendar", x=x, y=y, w=w, h=h, sort_order=sort_order, created_at=time.time()) db_session.add(widget) db_session.flush() db_session.add(CalendarWidgetConfig(widget_id=widget.id)) db_session.commit() return widget def _add_tasks_widget(db_session, frame_id=1, x=3, y=0, w=2, h=2, sort_order=2) -> Widget: widget = Widget(frame_id=frame_id, widget_type="tasks", x=x, y=y, w=w, h=h, sort_order=sort_order, created_at=time.time()) db_session.add(widget) db_session.flush() db_session.add(TaskWidgetConfig(widget_id=widget.id)) db_session.commit() return widget def _add_weather_widget(db_session, frame_id=1, x=0, y=0, w=2, h=2, sort_order=1) -> Widget: widget = Widget(frame_id=frame_id, widget_type="weather", x=x, y=y, w=w, h=h, sort_order=sort_order, created_at=time.time()) db_session.add(widget) db_session.flush() db_session.add(WeatherWidgetConfig(widget_id=widget.id)) db_session.commit() return widget def _setup_alice(client) -> None: resp = client.post("/setup", data={"username": "alice", "password": "hunter22"}) assert resp.status_code == 303, resp.text # --- save ------------------------------------------------------------------ def test_save_captures_placement_and_photo_settings_but_not_queue_state(client, db_session): _setup_alice(client) photos_id = _widget_id(db_session) with db_session.no_autoflush: pcfg = db_session.get(PhotoWidgetConfig, photos_id) pcfg.album_id = "album-1" pcfg.order = "shuffle" pcfg.queue_target_len = 30 pcfg.current_asset_id = "some-asset" pcfg.queue = ["a", "b", "c"] db_session.commit() resp = client.post("/api/frames/1/layouts", json={"name": "My Layout"}, headers=csrf_headers(client)) assert resp.status_code == 200, resp.text body = resp.json() assert body["name"] == "My Layout" assert body["cols"] == 8 and body["rows"] == 5 assert body["widget_count"] == 1 layout = db_session.query(SavedLayout).filter_by(user_id=1, name="My Layout").one() snap = db_session.query(SavedLayoutWidget).filter_by(saved_layout_id=layout.id).one() assert (snap.widget_type, snap.x, snap.y, snap.w, snap.h) == ("photos", 0, 0, 8, 5) assert snap.config == {"album_id": "album-1", "order": "shuffle", "display_mode": "crop_faces", "queue_target_len": 30} def test_save_and_apply_round_trip_weather_settings(client, db_session): _setup_alice(client) weather = _add_weather_widget(db_session) with db_session.no_autoflush: wcfg = db_session.get(WeatherWidgetConfig, weather.id) wcfg.mode = "daily" wcfg.provider = "nws" wcfg.units = "celsius" wcfg.city_label = "Boston, MA" wcfg.city_latitude = 42.36 wcfg.city_longitude = -71.06 wcfg.hourly_interval_hours = 6 wcfg.daily_days = 7 wcfg.checked_at = 12345.0 wcfg.cached = {"stale": "runtime state, not a setting"} db_session.commit() db_session.expunge(wcfg) save_resp = client.post("/api/frames/1/layouts", json={"name": "Weather Layout"}, headers=csrf_headers(client)) assert save_resp.status_code == 200, save_resp.text layout = db_session.query(SavedLayout).filter_by(user_id=1, name="Weather Layout").one() snap = db_session.query(SavedLayoutWidget).filter_by(saved_layout_id=layout.id, widget_type="weather").one() assert snap.config == { "mode": "daily", "provider": "nws", "units": "celsius", "city_label": "Boston, MA", "city_latitude": 42.36, "city_longitude": -71.06, "hourly_interval_hours": 6, "daily_days": 7, "cities": None, "render_style": "classic", } client.delete("/api/frames/1/widgets", headers=csrf_headers(client)) apply_resp = client.post(f"/api/frames/1/layouts/{layout.id}/apply", headers=csrf_headers(client)) assert apply_resp.status_code == 200, apply_resp.text new_widget = db_session.query(Widget).filter_by(frame_id=1, widget_type="weather").one() new_cfg = db_session.get(WeatherWidgetConfig, new_widget.id) assert (new_cfg.mode, new_cfg.provider, new_cfg.units) == ("daily", "nws", "celsius") assert (new_cfg.city_label, new_cfg.city_latitude, new_cfg.city_longitude) == ("Boston, MA", 42.36, -71.06) assert (new_cfg.hourly_interval_hours, new_cfg.daily_days) == (6, 7) # Runtime fetch-cache state is never captured/restored by a saved layout. assert new_cfg.checked_at == 0.0 assert new_cfg.cached is None def test_save_captures_calendar_sources_and_button_actions(client, db_session): _setup_alice(client) db_session.query(Widget).filter_by(frame_id=1, widget_type="photos").delete() db_session.commit() cal = _add_calendar_widget(db_session) db_session.add(FrameCalendar(widget_id=cal.id, user_id=1, calendar_key="ics", calendar_label="Alice", included=True, color_index=2)) db_session.add(FrameButtonAction(frame_id=1, button="next", widget_id=cal.id, action="advance", sort_order=0)) db_session.commit() resp = client.post("/api/frames/1/layouts", json={"name": "Calendar Layout"}, headers=csrf_headers(client)) assert resp.status_code == 200, resp.text layout = db_session.query(SavedLayout).filter_by(user_id=1, name="Calendar Layout").one() snap = db_session.query(SavedLayoutWidget).filter_by(saved_layout_id=layout.id).one() source = db_session.query(SavedLayoutSource).filter_by(saved_layout_widget_id=snap.id).one() assert (source.kind, source.user_id, source.calendar_key, source.color_index) == ("calendar", 1, "ics", 2) action = db_session.query(SavedLayoutButtonAction).filter_by(saved_layout_widget_id=snap.id).one() assert (action.button, action.action) == ("next", "advance") def test_save_with_existing_name_overwrites_in_place(client, db_session): _setup_alice(client) resp1 = client.post("/api/frames/1/layouts", json={"name": "Layout A"}, headers=csrf_headers(client)) layout_id = resp1.json()["id"] photos_id = _widget_id(db_session) client.patch(f"/api/frames/1/widgets/{photos_id}", json={"x": 0, "y": 0, "w": 4, "h": 5}, headers=csrf_headers(client)) resp2 = client.post("/api/frames/1/layouts", json={"name": "Layout A"}, headers=csrf_headers(client)) assert resp2.status_code == 200, resp2.text assert resp2.json()["id"] == layout_id assert db_session.query(SavedLayout).filter_by(user_id=1, name="Layout A").count() == 1 snap = db_session.query(SavedLayoutWidget).filter_by(saved_layout_id=layout_id).one() assert (snap.w, snap.h) == (4, 5) def test_save_rejects_blank_name(client, db_session): _setup_alice(client) resp = client.post("/api/frames/1/layouts", json={"name": " "}, headers=csrf_headers(client)) assert resp.status_code == 400 def test_save_linked_but_not_controlling_user_409s(client, db_session): _setup_alice(client) bob = make_user(db_session, "bob") link_user(db_session, bob, db_session.get(Frame, 1)) client.cookies.clear() login(client, "bob") resp = client.post("/api/frames/1/layouts", json={"name": "Bob's Layout"}, headers=csrf_headers(client)) assert resp.status_code == 409 assert resp.json()["detail"]["error"] == "not_controller" # --- list -------------------------------------------------------------- def test_list_is_scoped_to_the_calling_user(client, db_session): _setup_alice(client) bob = make_user(db_session, "bob") link_user(db_session, bob, db_session.get(Frame, 1)) client.post("/api/frames/1/layouts", json={"name": "Alice's Layout"}, headers=csrf_headers(client)) client.cookies.clear() login(client, "bob") resp = client.get("/api/frames/1/layouts") assert resp.status_code == 200 assert resp.json()["layouts"] == [] def test_list_flags_layouts_incompatible_with_this_frames_grid(client, db_session): _setup_alice(client) client.post("/api/frames/1/layouts", json={"name": "Landscape Layout"}, headers=csrf_headers(client)) other_frame = Frame(name="Portrait frame", manage_token="tok-2", device_id="dev-2", device_token="dtok-2", orientation="portrait") db_session.add(other_frame) db_session.commit() from app import grid from app.models import Widget as W cols, rows = grid.grid_dims("portrait") db_session.add(W(frame_id=other_frame.id, widget_type="photos", x=0, y=0, w=cols, h=rows, sort_order=0, created_at=time.time())) db_session.commit() resp = client.get(f"/api/frames/{other_frame.id}/layouts") assert resp.status_code == 200 layouts = resp.json()["layouts"] assert len(layouts) == 1 assert layouts[0]["name"] == "Landscape Layout" assert layouts[0]["compatible"] is False resp2 = client.get("/api/frames/1/layouts") assert resp2.json()["layouts"][0]["compatible"] is True # --- rename / delete ----------------------------------------------------- def test_rename_updates_the_name(client, db_session): _setup_alice(client) resp = client.post("/api/frames/1/layouts", json={"name": "Old Name"}, headers=csrf_headers(client)) layout_id = resp.json()["id"] rename_resp = client.patch(f"/api/layouts/{layout_id}", json={"name": "New Name"}, headers=csrf_headers(client)) assert rename_resp.status_code == 200, rename_resp.text assert rename_resp.json()["name"] == "New Name" assert db_session.get(SavedLayout, layout_id).name == "New Name" def test_rename_rejects_conflicting_name(client, db_session): _setup_alice(client) client.post("/api/frames/1/layouts", json={"name": "Layout A"}, headers=csrf_headers(client)) resp_b = client.post("/api/frames/1/layouts", json={"name": "Layout B"}, headers=csrf_headers(client)) layout_b_id = resp_b.json()["id"] resp = client.patch(f"/api/layouts/{layout_b_id}", json={"name": "Layout A"}, headers=csrf_headers(client)) assert resp.status_code == 400 def test_rename_someone_elses_layout_404s(client, db_session): _setup_alice(client) resp = client.post("/api/frames/1/layouts", json={"name": "Alice's Layout"}, headers=csrf_headers(client)) layout_id = resp.json()["id"] bob = make_user(db_session, "bob") link_user(db_session, bob, db_session.get(Frame, 1)) client.cookies.clear() login(client, "bob") resp = client.patch(f"/api/layouts/{layout_id}", json={"name": "Hijacked"}, headers=csrf_headers(client)) assert resp.status_code == 404 assert db_session.get(SavedLayout, layout_id).name == "Alice's Layout" def test_delete_removes_the_layout_and_cascades(client, db_session): _setup_alice(client) resp = client.post("/api/frames/1/layouts", json={"name": "Doomed"}, headers=csrf_headers(client)) layout_id = resp.json()["id"] del_resp = client.delete(f"/api/layouts/{layout_id}", headers=csrf_headers(client)) assert del_resp.status_code == 200 assert db_session.get(SavedLayout, layout_id) is None assert db_session.query(SavedLayoutWidget).filter_by(saved_layout_id=layout_id).count() == 0 def test_delete_someone_elses_layout_404s(client, db_session): _setup_alice(client) resp = client.post("/api/frames/1/layouts", json={"name": "Alice's Layout"}, headers=csrf_headers(client)) layout_id = resp.json()["id"] bob = make_user(db_session, "bob") link_user(db_session, bob, db_session.get(Frame, 1)) client.cookies.clear() login(client, "bob") resp = client.delete(f"/api/layouts/{layout_id}", headers=csrf_headers(client)) assert resp.status_code == 404 assert db_session.get(SavedLayout, layout_id) is not None # --- apply --------------------------------------------------------------- def test_apply_replaces_widgets_and_restores_sources_and_button_actions(client, db_session): _setup_alice(client) db_session.query(Widget).filter_by(frame_id=1, widget_type="photos").delete() db_session.commit() cal = _add_calendar_widget(db_session, x=0, y=0, w=3, h=2, sort_order=0) tasks = _add_tasks_widget(db_session, x=3, y=0, w=2, h=2, sort_order=1) db_session.add(FrameCalendar(widget_id=cal.id, user_id=1, calendar_key="ics", calendar_label="Alice", included=True)) db_session.add(FrameTaskList(widget_id=tasks.id, user_id=1, calendar_key="caldav:/tasks/", included=True)) db_session.add(FrameButtonAction(frame_id=1, button="next", widget_id=cal.id, action="advance", sort_order=0)) db_session.add(FrameButtonAction(frame_id=1, button="back", widget_id=cal.id, action="back", sort_order=0)) db_session.commit() save_resp = client.post("/api/frames/1/layouts", json={"name": "Rich Layout"}, headers=csrf_headers(client)) layout_id = save_resp.json()["id"] # Blow away the current arrangement so apply has something real to restore. client.delete("/api/frames/1/widgets", headers=csrf_headers(client)) assert db_session.query(Widget).filter_by(frame_id=1).count() == 0 apply_resp = client.post(f"/api/frames/1/layouts/{layout_id}/apply", headers=csrf_headers(client)) assert apply_resp.status_code == 200, apply_resp.text assert apply_resp.json()["widget_count"] == 2 widgets = db_session.query(Widget).filter_by(frame_id=1).order_by(Widget.sort_order).all() assert [w.widget_type for w in widgets] == ["calendar", "tasks"] new_cal, new_tasks = widgets cal_source = db_session.query(FrameCalendar).filter_by(widget_id=new_cal.id).one() assert (cal_source.user_id, cal_source.calendar_key) == (1, "ics") task_source = db_session.query(FrameTaskList).filter_by(widget_id=new_tasks.id).one() assert (task_source.user_id, task_source.calendar_key) == (1, "caldav:/tasks/") actions = db_session.query(FrameButtonAction).filter_by(frame_id=1).all() assert len(actions) == 2 assert all(a.widget_id == new_cal.id for a in actions) assert {a.button for a in actions} == {"next", "back"} def test_apply_rejects_mismatched_grid(client, db_session): _setup_alice(client) save_resp = client.post("/api/frames/1/layouts", json={"name": "Landscape Layout"}, headers=csrf_headers(client)) layout_id = save_resp.json()["id"] other_frame = Frame(name="Portrait frame", manage_token="tok-2", device_id="dev-2", device_token="dtok-2", orientation="portrait", controlled_by_user_id=1) db_session.add(other_frame) db_session.commit() from app import grid from app.models import Widget as W cols, rows = grid.grid_dims("portrait") db_session.add(W(frame_id=other_frame.id, widget_type="photos", x=0, y=0, w=cols, h=rows, sort_order=0, created_at=time.time())) db_session.commit() resp = client.post(f"/api/frames/{other_frame.id}/layouts/{layout_id}/apply", headers=csrf_headers(client)) assert resp.status_code == 400 assert "different frame size" in resp.json()["detail"].lower() def test_apply_drops_a_source_whose_owning_user_no_longer_exists(client, db_session): _setup_alice(client) db_session.query(Widget).filter_by(frame_id=1, widget_type="photos").delete() db_session.commit() cal = _add_calendar_widget(db_session) bob = make_user(db_session, "bob") link_user(db_session, bob, db_session.get(Frame, 1)) bob_id = bob.id db_session.add(FrameCalendar(widget_id=cal.id, user_id=bob_id, calendar_key="ics", calendar_label="Bob", included=True)) db_session.commit() save_resp = client.post("/api/frames/1/layouts", json={"name": "Shared Calendar"}, headers=csrf_headers(client)) layout_id = save_resp.json()["id"] db_session.query(User).filter_by(id=bob_id).delete() db_session.commit() client.delete("/api/frames/1/widgets", headers=csrf_headers(client)) apply_resp = client.post(f"/api/frames/1/layouts/{layout_id}/apply", headers=csrf_headers(client)) assert apply_resp.status_code == 200, apply_resp.text new_cal = db_session.query(Widget).filter_by(frame_id=1, widget_type="calendar").one() assert db_session.query(FrameCalendar).filter_by(widget_id=new_cal.id).count() == 0 def test_apply_someone_elses_layout_404s(client, db_session): _setup_alice(client) resp = client.post("/api/frames/1/layouts", json={"name": "Alice's Layout"}, headers=csrf_headers(client)) layout_id = resp.json()["id"] bob = make_user(db_session, "bob") link_user(db_session, bob, db_session.get(Frame, 1)) frame = db_session.get(Frame, 1) frame.controlled_by_user_id = bob.id db_session.commit() client.cookies.clear() login(client, "bob") resp = client.post(f"/api/frames/1/layouts/{layout_id}/apply", headers=csrf_headers(client)) assert resp.status_code == 404