pigway-frame-flow/tests/test_photo_resolution.py

import io

import pytest
from PIL import Image

from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput


def test_minimum_resolution_filters_catalog_and_persists(
    client, service, settings, legacy_photo_engine
):
    root = settings.data_dir.parent / "photos"
    root.mkdir()
    settings.photo_root = root
    for name, size in [
        ("small.jpg", (160, 120)),
        ("wide.jpg", (800, 600)),
        ("tall.jpg", (600, 800)),
    ]:
        Image.new("RGB", size).save(root / name)
    response = client.put(
        "/api/v1/photo-sources/family",
        json={
            "name": "Family",
            "kind": "local",
            "location": ".",
            "minimum_short_edge": 600,
            "minimum_long_edge": 800,
        },
    )
    assert response.status_code == 200
    assert response.json()["minimum_short_edge"] == 600
    assert service.photo_sources.get("family").minimum_long_edge == 800
    assert {x["name"] for x in service.photo_sources.catalog("family")} == {
        "wide.jpg",
        "tall.jpg",
    }
    response = client.put(
        "/api/v1/photo-sources/family",
        json={
            "name": "Family",
            "kind": "local",
            "location": ".",
        },
    )
    assert response.status_code == 200
    assert len(service.photo_sources.catalog("family")) == 3


def test_resolution_rejects_small_synology_preview(service, monkeypatch):
    photos = service.photo_sources
    photos.save(
        "family",
        PhotoSourceInput(
            name="Family",
            kind="synology-photos",
            location=".",
            minimum_short_edge=600,
            minimum_long_edge=800,
        ),
    )
    row = {"id": "1", "name": "photo.jpg"}
    monkeypatch.setattr(photos, "catalog", lambda *args, **kwargs: [row])
    photos.api_items["family"] = {"1": {}}
    monkeypatch.setattr(photos.synology, "item_people", lambda *a, **kw: [])

    def image(size):
        out = io.BytesIO()
        Image.new("RGB", size).save(out, "JPEG")
        return out.getvalue()

    monkeypatch.setattr(photos.synology, "download", lambda *a, **kw: image((300, 400)))
    with pytest.raises(FlowError, match="minimum resolution"):
        photos.read("family", "1")
    monkeypatch.setattr(photos.synology, "download", lambda *a, **kw: image((800, 600)))
    assert photos.read("family", "1")[1] == "photo.jpg"


def test_minimum_resolution_rejects_negative_value(client):
    response = client.put(
        "/api/v1/photo-sources/family",
        json={
            "name": "Family",
            "kind": "synology-photos",
            "location": ".",
            "minimum_short_edge": -1,
        },
    )
    assert response.status_code == 422


def test_synology_metadata_resolution_filters_before_catalog_limit(
    client, service, monkeypatch
):
    photos = service.photo_sources
    monkeypatch.setattr(photos.synology, "sid", lambda: "fixture")
    rows = [
        {
            "id": i,
            "filename": f"{i}.jpg",
            "additional": {
                "folder": "/family",
                "resolution": {"width": w, "height": h},
            },
        }
        for i, w, h in [(1, 300, 400), (2, 800, 600), (3, 0, 0)]
    ]
    monkeypatch.setattr(photos.synology, "items", lambda *a, **kw: rows)
    response = client.put(
        "/api/v1/photo-sources/family",
        json={
            "name": "Family",
            "kind": "synology-photos",
            "location": ".",
            "minimum_short_edge": 600,
            "minimum_long_edge": 800,
            "catalog_limit": 1,
        },
    )
    assert response.status_code == 200
    assert response.json()["minimum_long_edge"] == 800
    assert [x["id"] for x in photos.catalog("family")] == ["2"]