Files
espresso_frame/server/tests/test_manage_widget_dispatch.py
tfaour c323402895 Fix scan-to-download auth and share every photo widget's current photo
The share QR's URL carried no auth params at all, so it silently fell
back through require_device's legacy-token resolution to whichever
frame happened to still be flagged legacy -- working only by accident
for a single frame, sharing the wrong frame's photos for any other, and
going fully dead once that frame's legacy flag was cleared.

Move the endpoint to manage.py, keyed on the frame's own manage_token
(same pattern /m/<manage_token> already uses) instead of device auth.
Since the server now resolves assets itself instead of trusting a
caller-supplied asset_id, it naturally generalizes to gather every
photo widget's current photo into one Immich share link, not just one
"primary" widget's.
2026-07-27 14:38:26 +00:00

225 lines
9.0 KiB
Python

"""routers/manage.py -- the no-login "scan to manage" surface -- against
the widget-scoped photo state it was just repointed at. No prior test
coverage existed for this router at all before this file; it's
exercised here specifically because of how much its photo-queue
endpoints changed in the widget-system cutover (frame.current_asset_id/
frame.queue -> the frame's photo widget's own PhotoWidgetConfig)."""
from __future__ import annotations
from app.models import Frame, PhotoWidgetConfig
_ASSETS = [{"id": "asset-1"}, {"id": "asset-2"}, {"id": "asset-3"}]
def _mock_immich(monkeypatch):
"""manage.py imports immich_client_for/list_assets into its own
module namespace (`from .common import ...`) -- patching those
names, not app.widgets.photos' separate copies, since manage.py's
endpoints call its own bound names directly, never through
app/widgets/."""
monkeypatch.setattr("app.routers.manage.immich_client_for", lambda frame: object())
monkeypatch.setattr("app.routers.manage.list_assets", lambda client, album_id: _ASSETS)
def _configure_photo_widget(db_session):
from app.models import Widget
frame = db_session.get(Frame, 1)
# photo_widget_config_or_404 (see routers/common.py) checks Immich
# creds directly, not through the mocked immich_client_for below.
frame.immich_url = "http://immich.example.com"
frame.immich_api_key = "key"
widget = db_session.query(Widget).filter_by(frame_id=frame.id, widget_type="photos").one()
cfg = db_session.get(PhotoWidgetConfig, widget.id)
cfg.album_id = "album-1"
db_session.commit()
return frame, widget
def test_manage_queue_requires_configured_photo_widget(client, db_session):
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame = db_session.get(Frame, 1)
resp = client.get(f"/api/m/{frame.manage_token}/queue")
assert resp.status_code == 400
def test_manage_queue_returns_current_and_upcoming(client, db_session, monkeypatch):
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame, widget = _configure_photo_widget(db_session)
_mock_immich(monkeypatch)
resp = client.get(f"/api/m/{frame.manage_token}/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"]
def test_manage_advance_and_back(client, db_session, monkeypatch):
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame, widget = _configure_photo_widget(db_session)
_mock_immich(monkeypatch)
# establish a current photo first
client.get(f"/api/m/{frame.manage_token}/queue")
resp = client.post(f"/api/m/{frame.manage_token}/advance")
assert resp.status_code == 200
cfg = db_session.get(PhotoWidgetConfig, widget.id)
first_current = cfg.current_asset_id
assert first_current != ""
resp = client.post(f"/api/m/{frame.manage_token}/advance")
assert resp.status_code == 200
cfg = db_session.get(PhotoWidgetConfig, widget.id)
assert cfg.current_asset_id != first_current
resp = client.post(f"/api/m/{frame.manage_token}/back")
assert resp.status_code == 200
cfg = db_session.get(PhotoWidgetConfig, widget.id)
assert cfg.current_asset_id == first_current
def test_manage_promote(client, db_session, monkeypatch):
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame, widget = _configure_photo_widget(db_session)
_mock_immich(monkeypatch)
client.get(f"/api/m/{frame.manage_token}/queue") # populate the queue
cfg = db_session.get(PhotoWidgetConfig, widget.id)
assert "asset-3" in cfg.queue
resp = client.post(f"/api/m/{frame.manage_token}/promote", json={"asset_id": "asset-3"})
assert resp.status_code == 200, resp.text
cfg = db_session.get(PhotoWidgetConfig, widget.id)
assert cfg.queue[0] == "asset-3"
def test_manage_thumbnail_scoped_to_showing_or_queued(client, db_session, monkeypatch):
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame, widget = _configure_photo_widget(db_session)
_mock_immich(monkeypatch)
client.get(f"/api/m/{frame.manage_token}/queue")
resp = client.get(f"/api/m/{frame.manage_token}/thumbnail/not-on-this-frame")
assert resp.status_code == 404
monkeypatch.setattr(
"app.routers.manage.immich_client_for",
lambda frame: type("C", (), {
"download_asset_thumbnail": lambda self, asset_id: (b"jpegbytes", "image/jpeg"),
})(),
)
resp = client.get(f"/api/m/{frame.manage_token}/thumbnail/asset-1")
assert resp.status_code == 200
assert resp.content == b"jpegbytes"
def test_unknown_manage_token_404s(client, db_session):
resp = client.get("/api/m/not-a-real-token/queue")
assert resp.status_code == 404
def test_manage_share_requires_current_photo(client, db_session):
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame, _ = _configure_photo_widget(db_session)
resp = client.get(f"/frame/share/{frame.manage_token}", follow_redirects=False)
assert resp.status_code == 404
def test_manage_share_creates_link_and_redirects(client, db_session, monkeypatch):
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame, widget = _configure_photo_widget(db_session)
cfg = db_session.get(PhotoWidgetConfig, widget.id)
cfg.current_asset_id = "asset-1"
db_session.commit()
calls = []
monkeypatch.setattr(
"app.routers.manage.immich_client_for",
lambda frame: type("C", (), {
"create_share_link": lambda self, asset_ids, expires_in_s: (
calls.append(asset_ids) or "https://immich.example.com/share/abc"
),
})(),
)
resp = client.get(f"/frame/share/{frame.manage_token}", follow_redirects=False)
assert resp.status_code in (302, 303, 307)
assert calls == [["asset-1"]]
def test_manage_share_covers_every_photo_widget(client, db_session, monkeypatch):
from app.models import Widget
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame, widget = _configure_photo_widget(db_session)
cfg = db_session.get(PhotoWidgetConfig, widget.id)
cfg.current_asset_id = "asset-1"
second_widget = Widget(frame_id=frame.id, widget_type="photos", x=4, y=0, w=4, h=5,
sort_order=1, created_at=0)
db_session.add(second_widget)
db_session.flush()
db_session.add(PhotoWidgetConfig(widget_id=second_widget.id, album_id="album-2", current_asset_id="asset-9"))
db_session.commit()
calls = []
monkeypatch.setattr(
"app.routers.manage.immich_client_for",
lambda frame: type("C", (), {
"create_share_link": lambda self, asset_ids, expires_in_s: (
calls.append(asset_ids) or "https://immich.example.com/share/abc"
),
})(),
)
resp = client.get(f"/frame/share/{frame.manage_token}", follow_redirects=False)
assert resp.status_code in (302, 303, 307)
assert calls == [["asset-1", "asset-9"]]
def test_manage_share_scoped_to_its_own_frame(client, db_session, monkeypatch):
"""The regression test for the original bug: each frame's own share
QR must only ever surface that frame's own current photo, never
another frame's -- previously it silently resolved via whichever
frame happened to still carry the legacy migration token, regardless
of which frame's QR was actually scanned."""
from app.models import Widget
client.post("/setup", data={"username": "alice", "password": "hunter22"})
frame_a, widget_a = _configure_photo_widget(db_session)
cfg_a = db_session.get(PhotoWidgetConfig, widget_a.id)
cfg_a.current_asset_id = "asset-a"
frame_b = Frame(
name="Frame B", device_id="112233445566", device_token="devtok-b",
manage_token="mtok-b", orientation="landscape", created_at=0,
immich_url="http://immich.example.com", immich_api_key="key",
)
db_session.add(frame_b)
db_session.flush()
widget_b = Widget(frame_id=frame_b.id, widget_type="photos", x=0, y=0, w=4, h=5,
sort_order=0, created_at=0)
db_session.add(widget_b)
db_session.flush()
db_session.add(PhotoWidgetConfig(widget_id=widget_b.id, album_id="album-b", current_asset_id="asset-b"))
db_session.commit()
calls = []
monkeypatch.setattr(
"app.routers.manage.immich_client_for",
lambda frame: type("C", (), {
"create_share_link": lambda self, asset_ids, expires_in_s: (
calls.append(asset_ids) or "https://immich.example.com/share/abc"
),
})(),
)
client.get(f"/frame/share/{frame_a.manage_token}", follow_redirects=False)
client.get(f"/frame/share/{frame_b.manage_token}", follow_redirects=False)
assert calls == [["asset-a"], ["asset-b"]]
def test_unknown_manage_share_token_404s(client, db_session):
resp = client.get("/frame/share/not-a-real-token")
assert resp.status_code == 404