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"]