"""api_widgets.py's config-save (form-urlencoded, dispatched by widget_type) and the photo-queue endpoints it shares the file with -- the moved-and-consolidated counterparts of the old frame-level api_config_save/api_queue/etc. No prior HTTP-level coverage existed for either the old or new shape of these endpoints; added after manual browser testing caught api_widget_config_save expecting a JSON body while the (unmodified, copied-over) dialog JS posts form-urlencoded data -- a real bug an HTTP-level test would have caught immediately.""" from __future__ import annotations import io from PIL import Image from app.models import ( BatteryWidgetConfig, CalendarWidgetConfig, Frame, PhotoWidgetConfig, StaticWidgetConfig, TaskWidgetConfig, TextWidgetConfig, WeatherWidgetConfig, Widget, ) from .conftest import csrf_headers _ASSETS = [{"id": "asset-1"}, {"id": "asset-2"}, {"id": "asset-3"}] def _photo_widget(db_session) -> Widget: return db_session.query(Widget).filter_by(frame_id=1, widget_type="photos").one() def _add_calendar_widget(db_session) -> Widget: import time 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(widget) db_session.flush() db_session.add(CalendarWidgetConfig(widget_id=widget.id)) db_session.commit() return widget def _add_tasks_widget(db_session) -> Widget: import time widget = Widget(frame_id=1, widget_type="tasks", x=0, y=0, w=2, h=2, sort_order=1, 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_static_widget(db_session) -> Widget: import time widget = Widget(frame_id=1, widget_type="static", x=0, y=0, w=2, h=2, sort_order=1, created_at=time.time()) db_session.add(widget) db_session.flush() db_session.add(StaticWidgetConfig(widget_id=widget.id)) db_session.commit() return widget def _add_text_widget(db_session) -> Widget: import time widget = Widget(frame_id=1, widget_type="text", x=0, y=0, w=2, h=1, sort_order=1, created_at=time.time()) db_session.add(widget) db_session.flush() db_session.add(TextWidgetConfig(widget_id=widget.id)) db_session.commit() return widget def _add_weather_widget(db_session, **cfg_kwargs) -> Widget: import time widget = Widget(frame_id=1, widget_type="weather", x=0, y=0, w=2, h=2, sort_order=1, created_at=time.time()) db_session.add(widget) db_session.flush() db_session.add(WeatherWidgetConfig(widget_id=widget.id, **cfg_kwargs)) db_session.commit() return widget def _add_battery_widget(db_session, **cfg_kwargs) -> Widget: import time widget = Widget(frame_id=1, widget_type="battery", x=0, y=0, w=1, h=1, sort_order=1, created_at=time.time()) db_session.add(widget) db_session.flush() db_session.add(BatteryWidgetConfig(widget_id=widget.id, **cfg_kwargs)) db_session.commit() return widget def _png_bytes(size=(20, 10), color=(10, 20, 30)) -> bytes: buf = io.BytesIO() Image.new("RGB", size, color).save(buf, format="PNG") return buf.getvalue() def _mock_immich(monkeypatch): monkeypatch.setattr("app.routers.api_widgets.immich_client_for", lambda frame: object()) monkeypatch.setattr("app.routers.api_widgets.list_assets", lambda client, album_id: _ASSETS) # --- api_widget_config_save --- def test_config_save_updates_a_photos_widget(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _photo_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"album_id": "album-42", "order": "shuffle", "display_mode": "stretch_fill", "queue_target_len": "15"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(PhotoWidgetConfig, widget.id) assert cfg.album_id == "album-42" assert cfg.order == "shuffle" assert cfg.display_mode == "stretch_fill" assert cfg.queue_target_len == 15 def test_config_save_updates_a_calendar_widget(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_calendar_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"calendar_view": "week", "calendar_week_start": "1", "calendar_week_days": "5", "calendar_week_layout": "vertical", "calendar_weather_enabled": "true", "calendar_weather_units": "celsius"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(CalendarWidgetConfig, widget.id) assert cfg.view == "week" assert cfg.week_start == 1 assert cfg.week_days == 5 assert cfg.week_layout == "vertical" assert cfg.weather_enabled is True assert cfg.weather_units == "celsius" def test_config_save_updates_a_tasks_widget(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_tasks_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"tasks_name": "Chores", "tasks_show_completed": "true"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(TaskWidgetConfig, widget.id) assert cfg.name == "Chores" assert cfg.show_completed is True def test_config_save_updates_a_weather_widget(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_weather_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"weather_mode": "hourly", "weather_provider": "nws", "weather_units": "celsius", "weather_hourly_interval_hours": "6", "weather_daily_days": "3"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(WeatherWidgetConfig, widget.id) assert cfg.mode == "hourly" assert cfg.provider == "nws" assert cfg.units == "celsius" assert cfg.hourly_interval_hours == 6 assert cfg.daily_days == 3 def test_config_save_ignores_an_invalid_weather_mode(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_weather_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"weather_mode": "not_a_real_mode"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(WeatherWidgetConfig, widget.id) assert cfg.mode == "current" # unchanged -- invalid value silently ignored, same as calendar_view's own validation def test_config_save_switching_mode_clears_the_now_incompatible_cache(client, db_session, monkeypatch): """Regression test: a widget's `cached` shape depends on its mode (a single-temp dict for current, a list for hourly/multi_city, a dict for daily). Switching modes without clearing the old cache used to crash the very next preview -- get_or_refresh_weather_widget_data's multi_city branch tried `c["label"]` against a leftover "current"- mode dict, TypeError: string indices must be integers -- rather than just triggering a fresh fetch shaped for the new mode.""" client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_weather_widget(db_session, mode="current", cached={"temp": 70.0, "category": "clear"}, checked_at=1e15) # far in the future -- would still be "fresh" if not cleared resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"weather_mode": "multi_city"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(WeatherWidgetConfig, widget.id) assert cfg.cached is None cfg.cities = [{"label": "Portland, Oregon, United States", "latitude": 45.5, "longitude": -122.6}] db_session.commit() monkeypatch.setattr( "app.routers.api_widgets.get_or_refresh_weather_widget_data", lambda db, frame, widget, force=False: [ {"label": "Portland", "high": 75, "low": 55, "category": "clear"}, ], ) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/weather") assert resp.status_code == 200, resp.text def test_config_save_truncates_an_overlong_tasks_name(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_tasks_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"tasks_name": "A" * 200}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(TaskWidgetConfig, widget.id) assert len(cfg.name) == 40 def test_config_save_updates_a_static_widget(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"display_mode": "letterbox"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(StaticWidgetConfig, widget.id) assert cfg.display_mode == "letterbox" def test_config_save_rejects_an_unrecognized_static_display_mode(client, db_session): """Falls back to the default rather than erroring -- same "truncate/ clamp, don't reject" posture as the other config-save fields.""" client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"display_mode": "crop_faces"}, # valid for photos, not offered for static headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(StaticWidgetConfig, widget.id) assert cfg.display_mode == "crop_fill" def test_config_save_updates_a_battery_widget(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_battery_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"battery_mode": "compact"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(BatteryWidgetConfig, widget.id) assert cfg.mode == "compact" def test_config_save_rejects_an_unrecognized_battery_mode(client, db_session): """Falls back to the default rather than erroring -- same "clamp, don't reject" posture as the other config-save fields.""" client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_battery_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"battery_mode": "graph"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(BatteryWidgetConfig, widget.id) assert cfg.mode == "detailed" def test_config_save_updates_a_text_widget(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_text_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={ "text_html": '
Hello world
', "text_font_size": "40", "text_font_family": "serif", "text_align": "center", "text_background_color": "#ffdb00", }, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(TextWidgetConfig, widget.id) assert cfg.content == [[ {"text": "Hello ", "bold": False, "italic": False, "underline": False, "color": None, "bg": None}, {"text": "world", "bold": True, "italic": False, "underline": False, "color": None, "bg": None}, ]] assert cfg.font_size == 40 assert cfg.font_family == "serif" assert cfg.align == "center" assert cfg.background_color == "#ffdb00" def test_config_save_clamps_text_font_size_and_rejects_bad_family_align_and_color(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_text_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"text_font_size": "500", "text_font_family": "comic-sans", "text_align": "diagonal", "text_background_color": "not-a-color"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(TextWidgetConfig, widget.id) assert cfg.font_size == 96 # clamped to MAX_TEXT_FONT_SIZE assert cfg.font_family == "sans" # fell back to DEFAULT_FONT_FAMILY assert cfg.align == "left" # fell back to the default assert cfg.background_color == "#ffffff" # fell back to the default def test_config_save_only_partially_updates_provided_fields(client, db_session): """Fields not present in the POST are left untouched -- the whole point of the partial-update convention (each dialog's own form only ever posts its own fields).""" client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _photo_widget(db_session) cfg = db_session.get(PhotoWidgetConfig, widget.id) cfg.order = "shuffle" db_session.commit() resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"queue_target_len": "10"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(PhotoWidgetConfig, widget.id) assert cfg.queue_target_len == 10 assert cfg.order == "shuffle" # untouched def test_config_save_calendar_fields_are_a_no_op_on_a_photos_widget(client, db_session): """Posting calendar-shaped fields at a photos widget's config endpoint doesn't error -- it just doesn't apply, since the dispatch is purely by widget.widget_type.""" client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _photo_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"calendar_view": "week"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text def test_config_save_404s_for_a_widget_id_that_does_not_exist(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) resp = client.post("/api/frames/1/widgets/999999/config", data={"order": "shuffle"}, headers=csrf_headers(client)) assert resp.status_code == 404 # --- photo queue (moved from the old frame-level /api/frames/{id}/queue) --- def test_queue_requires_a_configured_album(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _photo_widget(db_session) resp = client.get(f"/api/frames/1/widgets/{widget.id}/queue") assert resp.status_code == 400 def test_queue_returns_current_and_upcoming(client, db_session, monkeypatch): client.post("/setup", data={"username": "alice", "password": "hunter22"}) frame = db_session.get(Frame, 1) frame.immich_url = "http://immich.example.com" frame.immich_api_key = "key" widget = _photo_widget(db_session) cfg = db_session.get(PhotoWidgetConfig, widget.id) cfg.album_id = "album-1" db_session.commit() _mock_immich(monkeypatch) resp = client.get(f"/api/frames/1/widgets/{widget.id}/queue") assert resp.status_code == 200, resp.text data = resp.json() assert data["current"]["id"] == "asset-1" assert [u["id"] for u in data["upcoming"]] == ["asset-2", "asset-3"] assert data["control"]["you"] is True def test_queue_400s_when_widget_is_not_photos(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_calendar_widget(db_session) resp = client.get(f"/api/frames/1/widgets/{widget.id}/queue") assert resp.status_code == 400 # --- static image: upload/preview ------------------------------------------ def test_static_upload_decodes_and_stores_the_image(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/static-upload", files={"file": ("photo.png", _png_bytes(size=(20, 10)), "image/png")}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text assert resp.json()["filename"] == "photo.png" cfg = db_session.get(StaticWidgetConfig, widget.id) assert cfg.image is not None assert Image.open(io.BytesIO(cfg.image)).size == (20, 10) assert cfg.original_filename == "photo.png" assert cfg.uploaded_at > 0 def test_static_upload_rejects_a_non_image_non_pdf_file(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/static-upload", files={"file": ("notes.txt", b"just some text", "text/plain")}, headers=csrf_headers(client), ) assert resp.status_code == 400 cfg = db_session.get(StaticWidgetConfig, widget.id) assert cfg.image is None def test_static_upload_400s_when_widget_is_not_static(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_calendar_widget(db_session) resp = client.post( f"/api/frames/1/widgets/{widget.id}/static-upload", files={"file": ("photo.png", _png_bytes(), "image/png")}, headers=csrf_headers(client), ) assert resp.status_code == 400 def test_preview_static_400s_before_anything_is_uploaded(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/static") assert resp.status_code == 400 def test_preview_static_renders_after_upload(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) client.post( f"/api/frames/1/widgets/{widget.id}/static-upload", files={"file": ("photo.png", _png_bytes(), "image/png")}, headers=csrf_headers(client), ) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/static") assert resp.status_code == 200, resp.text assert resp.headers["content-type"] == "image/png" # --- text: preview ----------------------------------------------------- def test_preview_text_400s_before_anything_is_authored(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_text_widget(db_session) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/text") assert resp.status_code == 400 def test_preview_text_renders_after_saving_content(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_text_widget(db_session) client.post( f"/api/frames/1/widgets/{widget.id}/config", data={"text_html": "
Hello world
"}, headers=csrf_headers(client), ) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/text") assert resp.status_code == 200, resp.text assert resp.headers["content-type"] == "image/png" def test_preview_text_400s_for_a_widget_that_is_not_text(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/text") assert resp.status_code == 400 # --- battery: preview --------------------------------------------------- def test_preview_battery_renders_even_before_any_report(client, db_session): """Unlike every other widget type's preview endpoint, there's no "not configured yet" 400 -- render() always has something to show (a placeholder, here, since the frame's never reported).""" client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_battery_widget(db_session) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/battery") assert resp.status_code == 200, resp.text assert resp.headers["content-type"] == "image/png" def test_preview_battery_renders_after_a_report(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_battery_widget(db_session) frame = db_session.get(Frame, 1) frame.battery_percent = 77 db_session.commit() resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/battery") assert resp.status_code == 200, resp.text assert resp.headers["content-type"] == "image/png" def test_preview_battery_400s_for_a_widget_that_is_not_battery(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/battery") assert resp.status_code == 400 # --- weather: location/cities/preview --------------------------------- def _mock_geocode(monkeypatch, label="Portland, Oregon, United States", latitude=45.5, longitude=-122.6): monkeypatch.setattr( "app.routers.api_widgets.weather.geocode_city", lambda name: {"label": label, "latitude": latitude, "longitude": longitude}, ) def test_weather_location_set_and_clear(client, db_session, monkeypatch): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_weather_widget(db_session, mode="current") _mock_geocode(monkeypatch) resp = client.post( f"/api/frames/1/widgets/{widget.id}/weather-location", json={"name": "Portland, OR"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text assert resp.json()["city"]["label"] == "Portland, Oregon, United States" cfg = db_session.get(WeatherWidgetConfig, widget.id) assert cfg.city_label == "Portland, Oregon, United States" assert cfg.city_latitude == 45.5 assert cfg.city_longitude == -122.6 resp = client.post( f"/api/frames/1/widgets/{widget.id}/weather-location", json={"name": None}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(WeatherWidgetConfig, widget.id) assert cfg.city_label is None assert cfg.city_latitude is None def test_weather_location_400s_for_a_widget_that_is_not_weather(client, db_session, monkeypatch): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_calendar_widget(db_session) _mock_geocode(monkeypatch) resp = client.post( f"/api/frames/1/widgets/{widget.id}/weather-location", json={"name": "Portland, OR"}, headers=csrf_headers(client), ) assert resp.status_code == 400 def test_weather_widget_cities_add_and_remove(client, db_session, monkeypatch): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_weather_widget(db_session, mode="multi_city") _mock_geocode(monkeypatch) resp = client.post( f"/api/frames/1/widgets/{widget.id}/weather-widget-cities/add", json={"name": "Portland, OR"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(WeatherWidgetConfig, widget.id) assert cfg.cities == [{"label": "Portland, Oregon, United States", "latitude": 45.5, "longitude": -122.6}] resp = client.post( f"/api/frames/1/widgets/{widget.id}/weather-widget-cities/add", json={"name": "Portland, OR"}, headers=csrf_headers(client), ) assert resp.status_code == 400 # already on the list resp = client.post( f"/api/frames/1/widgets/{widget.id}/weather-widget-cities/remove", json={"label": "Portland, Oregon, United States"}, headers=csrf_headers(client), ) assert resp.status_code == 200, resp.text cfg = db_session.get(WeatherWidgetConfig, widget.id) assert cfg.cities == [] def test_weather_widget_cities_400s_for_a_widget_that_is_not_weather(client, db_session, monkeypatch): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_calendar_widget(db_session) _mock_geocode(monkeypatch) resp = client.post( f"/api/frames/1/widgets/{widget.id}/weather-widget-cities/add", json={"name": "Portland, OR"}, headers=csrf_headers(client), ) assert resp.status_code == 400 def test_preview_weather_400s_before_anything_is_configured(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_weather_widget(db_session, mode="current") resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/weather") assert resp.status_code == 400 def test_preview_weather_400s_before_any_city_for_multi_city_mode(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_weather_widget(db_session, mode="multi_city") resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/weather") assert resp.status_code == 400 def test_preview_weather_renders_after_location_is_set(client, db_session, monkeypatch): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_weather_widget(db_session, mode="current") monkeypatch.setattr( "app.routers.api_widgets.get_or_refresh_weather_widget_data", lambda db, frame, widget, force=False: {"temp": 72.0, "category": "clear"}, ) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/weather") assert resp.status_code == 200, resp.text assert resp.headers["content-type"] == "image/png" def test_preview_weather_400s_for_a_widget_that_is_not_weather(client, db_session): client.post("/setup", data={"username": "alice", "password": "hunter22"}) widget = _add_static_widget(db_session) resp = client.get(f"/api/frames/1/widgets/{widget.id}/preview/weather") assert resp.status_code == 400