espresso_frame/server/tests/test_whiteboard_refresh_and_browse.py

"""get_or_refresh_whiteboard_for_widget's fetch throttle (and force=True
bypassing it) plus the whiteboard-browse HTTP endpoint (both now under
/api/frames/{id}/widgets/{widget_id}/...). 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 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 _add_bare_whiteboard_widget(db_session, frame: Frame) -> Widget:
    """A whiteboard widget with no source configured yet -- for tests
    that need the widget to exist (so the URL resolves) but want to
    exercise the "not configured" branches themselves."""
    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))
    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(f"/api/frames/1/widgets/{widget.id}/preview/whiteboard")
    assert resp.status_code == 200
    assert calls == []  # fresh cache, no force -- no refetch

    resp = client.get(f"/api/frames/1/widgets/{widget.id}/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"})
    frame = db_session.get(Frame, 1)
    widget = _add_bare_whiteboard_widget(db_session, frame)
    resp = client.get(f"/api/frames/1/widgets/{widget.id}/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()
    frame = db_session.get(Frame, 1)
    widget = _add_bare_whiteboard_widget(db_session, frame)

    resp = client.get(f"/api/frames/1/widgets/{widget.id}/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)
    widget = _add_bare_whiteboard_widget(db_session, 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(f"/api/frames/1/widgets/{widget.id}/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"