pigway-frame-flow/tests/test_thumbnail_plans.py

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
    )