pigway-frame-flow/tests/test_auto_collage.py

from io import BytesIO

import pytest
from PIL import Image
from pydantic import ValidationError

from pigway_frame_flow.models import AutoCollageContent


def photos(service, count=4):
    for i in range(count):
        data = BytesIO()
        Image.new("RGB", (320 + i, 200), (i * 50, 60, 100)).save(data, "JPEG")
        service.assets.put(data.getvalue(), f"photo-{i}.jpg")


def test_auto_preview_selects_exact_distinct_photos_without_frame_write(
    client, service
):
    photos(service)
    result = client.post(
        "/api/v1/render", json={"type": "auto_collage", "count": 3, "use_ai": False}
    )
    assert result.status_code == 202
    service.process_one()
    operation = service.operation(result.json()["operation_id"])
    assert operation["status"] == "COMPLETED"
    asset = operation["result"]["asset_id"]
    import json

    source = json.loads(
        service.db.one("SELECT payload FROM asset_sources WHERE asset_id=?", (asset,))[
            "payload"
        ]
    )
    regions = source["data"]["layout"]["regions"]
    assert len({x["asset_id"] for x in regions}) == 3
    assert service.registry.get("living-room").client.uploads == 0


def test_insufficient_photos_and_unconfigured_ai_fail_before_upload(service):
    from pigway_frame_flow.errors import FlowError

    photos(service, 1)
    for content in [
        AutoCollageContent(type="auto_collage", count=2, use_ai=False),
        AutoCollageContent(type="auto_collage", count=1),
    ]:
        with pytest.raises(FlowError):
            service.content_asset(content)
    assert service.registry.get("living-room").client.uploads == 0


@pytest.mark.parametrize("count", [0, 13])
def test_count_bounds(count):
    with pytest.raises(ValidationError):
        AutoCollageContent(type="auto_collage", count=count)


def test_auto_collage_job_contract_and_preview(client, service):
    photos(service, 2)
    job = client.post(
        "/api/v1/jobs",
        json={
            "name": "Collage",
            "device_id": "living-room",
            "enabled": False,
            "schedule": {"type": "daily", "time": "08:00"},
            "source": {"type": "auto_collage", "count": 2, "use_ai": False},
        },
    )
    assert job.status_code == 201
    result = client.post("/api/v1/jobs/" + job.json()["id"] + "/preview")
    assert result.status_code == 202
    service.process_one()
    assert service.operation(result.json()["operation_id"])["status"] == "COMPLETED"
    assert service.registry.get("living-room").client.uploads == 0


def test_source_selection_respects_quality_and_keeps_inputs_out_of_manual_library(
    service, monkeypatch
):
    from types import SimpleNamespace

    source = SimpleNamespace(
        minimum_sharpness=0,
        minimum_brightness=50,
        maximum_brightness=255,
        avoid_duplicates=False,
    )
    monkeypatch.setattr(service.photo_sources, "get", lambda _: source)
    monkeypatch.setattr(
        service.photo_sources, "catalog", lambda _: [{"id": str(i)} for i in range(3)]
    )

    def read(_, item):
        data = BytesIO()
        Image.new("RGB", (320, 240), (int(item) * 80,) * 3).save(data, "JPEG")
        return data.getvalue(), item + ".jpg", None

    monkeypatch.setattr(service.photo_sources, "read", read)
    content = AutoCollageContent(
        type="auto_collage", source_id="nas", count=2, use_ai=False
    )
    for _ in range(2):
        asset = service.content_asset(content)
        assert asset["source"] == "generated"
    assert (
        service.db.one("SELECT COUNT(*) AS n FROM assets WHERE kind='images'")["n"] == 0
    )
    assert (
        service.db.one("SELECT COUNT(*) AS n FROM assets WHERE kind='cache'")["n"] == 4
    )


def test_ai_receives_only_selected_asset_metadata(service, monkeypatch):
    from pigway_frame_flow.extensions.layout import LayoutStudio, local_plan

    photos(service, 3)
    calls = []

    def plan(_, request):
        calls.append(request)
        assert request.use_ai is True
        assert len(set(request.asset_ids)) == 3
        return local_plan(request)

    monkeypatch.setattr(LayoutStudio, "plan", plan)
    service.content_asset(
        AutoCollageContent(type="auto_collage", count=3, title="Family")
    )
    assert len(calls) == 1
    assert calls[0].title == "Family"
    assert service.registry.get("living-room").client.uploads == 0


def test_scheduled_collage_executes_once_and_tracks_cloud_task_head(app, service):
    from datetime import UTC, datetime, timedelta

    from pigway_frame_flow.models import JobInput

    photos(service, 3)
    scheduler = app.state.scheduler
    job = scheduler.save_job(
        JobInput(
            name="Scheduled collage",
            schedule={"type": "daily", "time": "08:00"},
            source={"type": "auto_collage", "count": 3, "use_ai": False},
        )
    )
    due = (datetime.now(UTC) - timedelta(seconds=2)).isoformat()
    service.db.execute("UPDATE jobs SET next_run=? WHERE id=?", (due, job["id"]))
    scheduler.tick()
    scheduler.tick()
    assert len(service.db.all("SELECT * FROM job_runs")) == 1
    service.process_one()
    service.sync()
    assert service.registry.get("living-room").client.uploads == 1
    row = service.db.one("SELECT job_id FROM managed_cloud_uploads")
    assert row["job_id"] == job["id"]