107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
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()
|