Make collage selection automatic and restore dashboard navigation order

tests/test_auto_collage.py

@@ -0,0 +1,160 @@
+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"]