feat: generate source-only watch thumbnails after photo uploads

tests/test_wrist_thumbnail.py

@@ -0,0 +1,106 @@
+import json
+from io import BytesIO
+
+import pytest
+from PIL import Image
+
+from pigway_frame_flow.errors import ErrorCode, FlowError
+from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
+
+
+@pytest.fixture
+def photo(service, settings, legacy_photo_engine):
+    root = settings.data_dir.parent / "photos"
+    root.mkdir()
+    settings.photo_root = root
+    Image.new("RGB", (900, 300), (30, 150, 220)).save(root / "photo.jpg")
+    service.photo_sources.save(
+        "family", PhotoSourceInput(name="Family", kind="local", location=".")
+    )
+
+
+def upload(service):
+    operation = service.submit(
+        "display",
+        {
+            "content": {"type": "photo_source", "source_id": "family"},
+            "options": {"priority": 0},
+        },
+        "living-room",
+        "test",
+        "request",
+    )
+    assert service.process_one()
+    return service.operation(operation)
+
+
+def test_uploaded_source_generates_full_bleed_watch_image(service, photo, monkeypatch):
+    # Deliberately decorate the frame rendition; the watch must use source pixels.
+    def captioned(*args, **kwargs):
+        output = BytesIO()
+        Image.new("RGB", kwargs["size"], "black").save(output, "JPEG")
+        return output.getvalue()
+
+    read = service.photo_sources.read
+    monkeypatch.setattr(
+        service.photo_sources,
+        "read",
+        lambda *a, **kw: (*read(*a, **kw)[:2], {"title": "Caption"}),
+    )
+    monkeypatch.setattr(
+        "pigway_frame_flow.display.service.render_captioned_photo", captioned
+    )
+    op = upload(service)
+    thumbnail = op["result"]["wrist_thumbnail"]
+    with Image.open(service.assets.path(thumbnail["asset_id"])) as image:
+        assert image.size == (410, 502)
+        assert image.mode == "RGB"
+        assert not image.info.get("progressive")
+        for point in [(0, 0), (409, 501), (205, 251)]:
+            assert image.getpixel(point)[2] > 200
+    row = service.db.one(
+        "SELECT payload FROM app_config WHERE key='wrist-thumbnail:living-room'"
+    )
+    assert json.loads(row["payload"]) == thumbnail
+
+
+def test_failed_upload_does_not_publish_thumbnail(service, photo, monkeypatch):
+    def fail(*args):
+        raise FlowError(ErrorCode.UPLOAD_FAILED, "Rejected", 502)
+
+    monkeypatch.setattr(service.registry.get("living-room"), "upload", fail)
+    op = upload(service)
+    assert op["status"] == "FAILED"
+    assert (
+        service.db.one(
+            "SELECT payload FROM app_config WHERE key LIKE 'wrist-thumbnail:%'"
+        )
+        is None
+    )
+
+
+def test_thumbnail_failure_does_not_change_upload_outcome(service, photo, monkeypatch):
+    def fail(*args, **kwargs):
+        raise OSError("Image encoder unavailable")
+
+    monkeypatch.setattr("pigway_frame_flow.display.wrist_thumbnail.convert", fail)
+    op = upload(service)
+    assert op["status"] == "SYNCING"
+    assert "wrist_thumbnail" not in op["result"]
+
+
+def test_local_upload_publishes_and_retention_protects_latest(service, photo):
+    from pigway_frame_flow.extensions.storage import LocalRetention
+
+    operation = service.submit(
+        "local-upload",
+        {"type": "photo_source", "source_id": "family"},
+        "living-room",
+        "test",
+        "local-request",
+    )
+    assert service.process_one()
+    op = service.operation(operation)
+    assert op["status"] == "COMPLETED"
+    asset_id = op["result"]["wrist_thumbnail"]["asset_id"]
+    assert asset_id in LocalRetention(service).protected()