tfaour
37bd657299
device.py's mode-keyed dispatch is replaced by a real compositor: load a frame's widgets, compute pixel rects via app/grid.py, render each through its widget module, and composite with render_panel. Physical NEXT/BACK buttons now execute each frame's assigned FrameButtonAction rows instead of one hardcoded per-mode action. api_frames.py, manage.py, and common.py's build_manage_content are repointed to read/write the frame's widget config rows instead of the old Frame columns, and every settings page (Photos/Calendar/ Whiteboard tabs) now pre-fills its form from the same widget config the write endpoints actually save to -- previously the read and write sides would have silently diverged. The old mode selector and photo-inlay checkbox are removed along with their now-inert wiring; arbitrary widget placement subsumes what the fixed inlay split did. Ships together with Phase 1 (per-type render/action modules) since splitting the read/write cutover across deploys would have left settings changes with no visible effect.
164 lines
6.5 KiB
Python
164 lines
6.5 KiB
Python
"""get_or_refresh_whiteboard_for_widget's fetch throttle (and force=True
|
|
bypassing it) plus the whiteboard-browse HTTP endpoint.
|
|
whiteboard.fetch_and_render is monkeypatched -- it talks to a real
|
|
WebDAV server and the Node render sidecar (see render-service/), neither
|
|
of which this suite needs a real copy of to verify the *throttle*/
|
|
*permission* logic around it."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from app import whiteboard
|
|
from app.models import Frame, User, Widget, WhiteboardWidgetConfig
|
|
|
|
from .conftest import csrf_headers, link_user, login, make_user
|
|
|
|
|
|
def _configure_whiteboard(db_session, frame: Frame, user: User) -> Widget:
|
|
widget = Widget(frame_id=frame.id, widget_type="whiteboard", x=0, y=0, w=8, h=5,
|
|
sort_order=1, created_at=time.time())
|
|
db_session.add(widget)
|
|
db_session.flush()
|
|
db_session.add(WhiteboardWidgetConfig(
|
|
widget_id=widget.id, user_id=user.id, url="http://example.invalid/board.whiteboard",
|
|
cached_image=b"OLD_CACHED_PNG", checked_at=time.time(), # just refreshed -- well within the throttle
|
|
))
|
|
db_session.commit()
|
|
return widget
|
|
|
|
|
|
def test_unforced_call_within_throttle_uses_cache(client, db_session, monkeypatch):
|
|
from app.routers.common import get_or_refresh_whiteboard_for_widget
|
|
|
|
client.post("/setup", data={"username": "alice", "password": "hunter22"})
|
|
alice = db_session.query(User).filter_by(username="alice").one()
|
|
alice.webdav_username = "alice"
|
|
alice.webdav_password = "secret"
|
|
frame = db_session.get(Frame, 1)
|
|
widget = _configure_whiteboard(db_session, frame, alice)
|
|
|
|
calls = []
|
|
monkeypatch.setattr(whiteboard, "fetch_and_render",
|
|
lambda url, u, p: calls.append(1) or b"NEW_PNG")
|
|
|
|
result = get_or_refresh_whiteboard_for_widget(db_session, frame, widget)
|
|
assert result == b"OLD_CACHED_PNG"
|
|
assert calls == []
|
|
|
|
|
|
def test_force_bypasses_throttle_and_persists(client, db_session, monkeypatch):
|
|
from app.routers.common import get_or_refresh_whiteboard_for_widget
|
|
|
|
client.post("/setup", data={"username": "alice", "password": "hunter22"})
|
|
alice = db_session.query(User).filter_by(username="alice").one()
|
|
alice.webdav_username = "alice"
|
|
alice.webdav_password = "secret"
|
|
frame = db_session.get(Frame, 1)
|
|
widget = _configure_whiteboard(db_session, frame, alice)
|
|
|
|
calls = []
|
|
monkeypatch.setattr(whiteboard, "fetch_and_render",
|
|
lambda url, u, p: calls.append(1) or b"NEW_PNG")
|
|
|
|
result = get_or_refresh_whiteboard_for_widget(db_session, frame, widget, force=True)
|
|
assert result == b"NEW_PNG"
|
|
assert len(calls) == 1
|
|
|
|
cfg = db_session.get(WhiteboardWidgetConfig, widget.id)
|
|
assert cfg.cached_image == b"NEW_PNG"
|
|
|
|
|
|
def test_preview_endpoint_force_query_param_bypasses_throttle(client, db_session, monkeypatch):
|
|
client.post("/setup", data={"username": "alice", "password": "hunter22"})
|
|
alice = db_session.query(User).filter_by(username="alice").one()
|
|
alice.webdav_username = "alice"
|
|
alice.webdav_password = "secret"
|
|
frame = db_session.get(Frame, 1)
|
|
|
|
# The preview endpoint runs whatever get_or_refresh_whiteboard_for_widget
|
|
# returns through PIL (Image.open) -- unlike the other tests in this
|
|
# file, the placeholder "cached" bytes need to be real, valid PNG
|
|
# data, not just an arbitrary marker string.
|
|
import io
|
|
|
|
from PIL import Image
|
|
buf = io.BytesIO()
|
|
Image.new("RGB", (1, 1), (255, 255, 255)).save(buf, format="PNG")
|
|
tiny_png = buf.getvalue()
|
|
|
|
widget = Widget(frame_id=frame.id, widget_type="whiteboard", x=0, y=0, w=8, h=5,
|
|
sort_order=1, created_at=time.time())
|
|
db_session.add(widget)
|
|
db_session.flush()
|
|
db_session.add(WhiteboardWidgetConfig(
|
|
widget_id=widget.id, user_id=alice.id, url="http://example.invalid/board.whiteboard",
|
|
cached_image=tiny_png, checked_at=time.time(),
|
|
))
|
|
db_session.commit()
|
|
|
|
calls = []
|
|
monkeypatch.setattr(whiteboard, "fetch_and_render",
|
|
lambda url, u, p: calls.append(1) or tiny_png)
|
|
|
|
resp = client.get("/api/frames/1/preview/whiteboard")
|
|
assert resp.status_code == 200
|
|
assert calls == [] # fresh cache, no force -- no refetch
|
|
|
|
resp = client.get("/api/frames/1/preview/whiteboard?force=1")
|
|
assert resp.status_code == 200
|
|
assert len(calls) == 1 # force=1 -- refetched despite fresh cache
|
|
|
|
|
|
# --- whiteboard-browse ---
|
|
|
|
def test_browse_requires_webdav_creds(client, db_session):
|
|
client.post("/setup", data={"username": "alice", "password": "hunter22"})
|
|
resp = client.get("/api/frames/1/whiteboard-browse")
|
|
assert resp.status_code == 400
|
|
assert "credentials" in resp.json()["detail"].lower()
|
|
|
|
|
|
def test_browse_requires_a_base_url_when_none_supplied(client, db_session):
|
|
client.post("/setup", data={"username": "alice", "password": "hunter22"})
|
|
alice = db_session.query(User).filter_by(username="alice").one()
|
|
alice.webdav_username = "alice"
|
|
alice.webdav_password = "secret"
|
|
db_session.commit()
|
|
|
|
resp = client.get("/api/frames/1/whiteboard-browse")
|
|
assert resp.status_code == 400
|
|
assert "browse root" in resp.json()["detail"].lower()
|
|
|
|
|
|
def test_browse_uses_the_caller_own_creds_not_the_frames_owner(client, db_session, monkeypatch):
|
|
"""bob, linked but not the frame's whiteboard owner, should still be
|
|
able to browse HIS OWN webdav account to pick a file for himself --
|
|
browsing is a "help me find a file" lookup against the caller's own
|
|
credentials, unrelated to whose account the frame is currently
|
|
pointed at."""
|
|
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)
|
|
bob.webdav_username = "bob"
|
|
bob.webdav_password = "bobsecret"
|
|
bob.webdav_base_url = "https://cloud.example.com/dav/files/bob/"
|
|
db_session.commit()
|
|
|
|
seen_creds = []
|
|
|
|
def fake_list_directory(url, username, password):
|
|
seen_creds.append((url, username, password))
|
|
return [{"name": "Board.whiteboard", "url": url + "Board.whiteboard", "is_dir": False}]
|
|
|
|
from app import webdav_client
|
|
monkeypatch.setattr(webdav_client, "list_directory", fake_list_directory)
|
|
|
|
client.cookies.clear()
|
|
login(client, "bob")
|
|
resp = client.get("/api/frames/1/whiteboard-browse")
|
|
assert resp.status_code == 200, resp.text
|
|
assert seen_creds == [("https://cloud.example.com/dav/files/bob/", "bob", "bobsecret")]
|
|
assert resp.json()["entries"][0]["name"] == "Board.whiteboard"
|