213 lines
7.3 KiB
Python
213 lines
7.3 KiB
Python
import json
|
|
from datetime import UTC, datetime
|
|
from io import BytesIO
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from pigway_frame_flow.models import JobInput
|
|
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
|
|
|
|
|
|
@pytest.fixture
|
|
def source(service, settings, legacy_photo_engine):
|
|
root = settings.data_dir.parent / "photos"
|
|
root.mkdir()
|
|
settings.photo_root = root
|
|
image = Image.new("RGB", (900, 300), "red")
|
|
image.paste("blue", (450, 0, 900, 300))
|
|
image.save(root / "photo.jpg")
|
|
service.photo_sources.save(
|
|
"family", PhotoSourceInput(name="Family", kind="local", location=".")
|
|
)
|
|
return {"source_id": "family", "item_id": "photo.jpg"}
|
|
|
|
|
|
def test_plan_crud_and_parameter_validation(client):
|
|
assert client.get("/ui/thumbnails.js").status_code == 200
|
|
plan = client.post("/api/v1/thumbnail-plans", json={"name": "Watch"}).json()
|
|
path = "/api/v1/thumbnail-plans/" + plan["id"]
|
|
assert client.get(path).json()["width"] == 410
|
|
assert len(client.get("/api/v1/thumbnail-plans").json()) == 1
|
|
assert (
|
|
client.put(
|
|
path, json={"name": "Square", "width": 240, "height": 240}
|
|
).status_code
|
|
== 200
|
|
)
|
|
for changes in [
|
|
{"width": 0},
|
|
{"height": 10000},
|
|
{"center_x": 2},
|
|
{"quality": 100},
|
|
{"fit": "stretch"},
|
|
{"extra": True},
|
|
]:
|
|
assert client.put(path, json={"name": "Invalid", **changes}).status_code == 422
|
|
assert client.delete(path).status_code == 204
|
|
assert client.get(path).status_code == 404
|
|
|
|
|
|
def test_preview_matches_generated_pixels_and_crop(client, source):
|
|
request = {
|
|
**source,
|
|
"plan": {"name": "Right", "width": 120, "height": 180, "center_x": 1},
|
|
}
|
|
preview = client.post("/api/v1/thumbnails/preview", json=request)
|
|
assert preview.status_code == 200
|
|
assert preview.headers["cache-control"] == "no-store"
|
|
asset = client.post("/api/v1/thumbnails/generate", json=request).json()
|
|
assert client.get(asset["url"]).content == preview.content
|
|
with Image.open(BytesIO(preview.content)) as image:
|
|
assert image.size == (120, 180)
|
|
assert image.getpixel((60, 90))[2] > 240
|
|
|
|
|
|
def test_task_requires_one_existing_plan_and_blocks_delete(client, app, source):
|
|
plan = client.post("/api/v1/thumbnail-plans", json={"name": "Watch"}).json()
|
|
job = {
|
|
"name": "Photos",
|
|
"schedule": {"type": "daily", "time": "07:00"},
|
|
"source": {"type": "photo_source", "source_id": source["source_id"]},
|
|
"generate_thumbnail": True,
|
|
"thumbnail_plan_id": plan["id"],
|
|
}
|
|
saved = client.post("/api/v1/jobs", json=job)
|
|
assert saved.status_code == 201
|
|
assert saved.json()["thumbnail_plan_id"] == plan["id"]
|
|
service = app.state.service
|
|
now = datetime.now(UTC)
|
|
service.db.execute(
|
|
"UPDATE jobs SET next_run=? WHERE id=?", (now.isoformat(), saved.json()["id"])
|
|
)
|
|
app.state.scheduler.tick(now)
|
|
pending = service.db.one("SELECT id,payload FROM operations WHERE action='display'")
|
|
assert json.loads(pending["payload"])["options"]["thumbnail_plan_id"] == plan["id"]
|
|
assert service.process_one()
|
|
assert (
|
|
service.operation(pending["id"])["result"]["wrist_thumbnail"]["plan_id"]
|
|
== plan["id"]
|
|
)
|
|
assert client.delete("/api/v1/thumbnail-plans/" + plan["id"]).status_code == 409
|
|
assert (
|
|
client.post("/api/v1/jobs", json={**job, "thumbnail_plan_id": None}).status_code
|
|
== 422
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/v1/jobs", json={**job, "thumbnail_plan_id": "0" * 32}
|
|
).status_code
|
|
== 404
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/v1/jobs",
|
|
json={**job, "source": {"type": "template", "template": "quote"}},
|
|
).status_code
|
|
== 422
|
|
)
|
|
assert (
|
|
JobInput.model_validate(
|
|
{
|
|
k: v
|
|
for k, v in job.items()
|
|
if k not in ("generate_thumbnail", "thumbnail_plan_id")
|
|
}
|
|
).generate_thumbnail
|
|
is False
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("enabled", [True, False])
|
|
def test_upload_uses_selected_plan_or_skips(service, client, source, enabled):
|
|
plan = client.post(
|
|
"/api/v1/thumbnail-plans", json={"name": "Square", "width": 128, "height": 128}
|
|
).json()
|
|
operation = service.submit(
|
|
"display",
|
|
{
|
|
"content": {"type": "photo_source", "source_id": "family"},
|
|
"options": {
|
|
"priority": 0,
|
|
"generate_thumbnail": enabled,
|
|
"thumbnail_plan_id": plan["id"] if enabled else None,
|
|
},
|
|
},
|
|
"living-room",
|
|
"test",
|
|
"request",
|
|
)
|
|
service.process_one()
|
|
result = service.operation(operation)["result"]
|
|
if enabled:
|
|
assert result["wrist_thumbnail"]["width"] == 128
|
|
assert result["wrist_thumbnail"]["plan_id"] == plan["id"]
|
|
else:
|
|
assert "wrist_thumbnail" not in result
|
|
|
|
|
|
def test_watch_gallery_separates_thumbnails_and_deletes_latest(client, service, source):
|
|
original = service.assets.put(b"fixture", "normal.jpg", "images")
|
|
generated = client.post(
|
|
"/api/v1/thumbnails/generate", json={**source, "plan": {"name": "Watch"}}
|
|
).json()
|
|
operation = service.submit(
|
|
"display",
|
|
{
|
|
"content": {"type": "photo_source", "source_id": "family"},
|
|
"options": {"priority": 0},
|
|
},
|
|
"living-room",
|
|
"test",
|
|
"request",
|
|
)
|
|
service.process_one()
|
|
auto = service.operation(operation)["result"]["wrist_thumbnail"]
|
|
page = client.get("/api/v1/thumbnails?limit=1").json()
|
|
assert page["total"] == 2 and len(page["items"]) == 1
|
|
items = client.get("/api/v1/thumbnails").json()["items"]
|
|
assert {x["asset_id"] for x in items} == {generated["asset_id"], auto["asset_id"]}
|
|
assert all(x["thumbnail"]["target_id"] == "watch" for x in items)
|
|
assert {x["asset_id"] for x in client.get("/api/v1/assets").json()} == {
|
|
original["asset_id"]
|
|
}
|
|
assert (
|
|
client.get("/api/v1/thumbnail-devices").json()[0]["transport_connected"]
|
|
is False
|
|
)
|
|
assert client.delete("/api/v1/assets/" + auto["asset_id"]).status_code == 204
|
|
assert client.get("/api/v1/thumbnails").json()["total"] == 1
|
|
assert (
|
|
service.db.one(
|
|
"SELECT payload FROM app_config WHERE key='wrist-thumbnail:living-room'"
|
|
)
|
|
is None
|
|
)
|
|
assert client.get(original["url"]).status_code == 200
|
|
|
|
|
|
def test_recover_legacy_receipt_does_not_guess_normal_images(service):
|
|
from pigway_frame_flow.display.thumbnail_gallery import recover
|
|
from pigway_frame_flow.storage.database import encode
|
|
|
|
thumbnail = service.assets.put(b"fixture", "watch.jpg", "cache")
|
|
normal = service.assets.put(b"fixture", "watch.jpg", "images")
|
|
service.db.execute(
|
|
"INSERT INTO app_config VALUES(?,?)",
|
|
(
|
|
"wrist-thumbnail:living-room",
|
|
encode({**thumbnail, "width": 410, "height": 502}),
|
|
),
|
|
)
|
|
recover(service)
|
|
recover(service)
|
|
assert service.db.one(
|
|
"SELECT payload FROM asset_sources WHERE asset_id=?", (thumbnail["asset_id"],)
|
|
)
|
|
assert (
|
|
service.db.one(
|
|
"SELECT payload FROM asset_sources WHERE asset_id=?", (normal["asset_id"],)
|
|
)
|
|
is None
|
|
)
|