109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
import pytest
|
|
from PIL import Image
|
|
|
|
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
|
|
|
|
|
|
def setup_source(service, settings):
|
|
root = settings.data_dir.parent / "photos"
|
|
root.mkdir()
|
|
settings.photo_root = root
|
|
for i in range(6):
|
|
Image.new("RGB", (300, 200)).save(root / f"{i}.jpg")
|
|
service.photo_sources.save(
|
|
"family", PhotoSourceInput(name="Family", kind="local", location=".")
|
|
)
|
|
return root
|
|
|
|
|
|
def test_pixel_limit_excludes_original_and_uses_preview(service, settings, monkeypatch):
|
|
root = setup_source(service, settings)
|
|
monkeypatch.setattr(
|
|
"pigway_frame_flow.providers.photo_sources.dimensions",
|
|
lambda *args: (8000, 4000),
|
|
)
|
|
assert service.photo_sources.catalog("family") == []
|
|
preview = root / "@eaDir" / "0.jpg"
|
|
preview.mkdir(parents=True)
|
|
Image.new("RGB", (100, 80)).save(preview / "SYNOPHOTO_THUMB_XL.jpg")
|
|
service.photo_sources.save(
|
|
"family",
|
|
PhotoSourceInput(
|
|
name="Family", kind="local", location=".", synology_previews=True
|
|
),
|
|
)
|
|
items = service.photo_sources.catalog("family")
|
|
assert len(items) == 1 and items[0]["provider_preview"]
|
|
|
|
|
|
def test_random_skips_invalid_photo_and_records_attempts(
|
|
client, service, settings, monkeypatch
|
|
):
|
|
setup_source(service, settings)
|
|
read = service.photo_sources.read
|
|
calls = []
|
|
|
|
def flaky(*args, **kwargs):
|
|
calls.append(args[1])
|
|
return (
|
|
(b"broken", "bad.jpg", None) if len(calls) == 1 else read(*args, **kwargs)
|
|
)
|
|
|
|
monkeypatch.setattr(service.photo_sources, "read", flaky)
|
|
response = client.post(
|
|
"/api/v1/devices/living-room/upload",
|
|
json={"type": "photo_source", "source_id": "family"},
|
|
)
|
|
service.process_one()
|
|
op = client.get("/api/v1/operations/" + response.json()["operation_id"]).json()
|
|
assert op["status"] == "COMPLETED"
|
|
assert [x["status"] for x in op["result"]["photo_attempts"]] == [
|
|
"SKIPPED",
|
|
"PREPARED",
|
|
]
|
|
assert len(set(calls)) == 2
|
|
|
|
|
|
def test_explicit_invalid_photo_never_changes_selection(
|
|
client, service, settings, monkeypatch
|
|
):
|
|
setup_source(service, settings)
|
|
monkeypatch.setattr(
|
|
service.photo_sources,
|
|
"read",
|
|
lambda *args, **kwargs: (b"broken", "bad.jpg", None),
|
|
)
|
|
response = client.post(
|
|
"/api/v1/devices/living-room/upload",
|
|
json={"type": "photo_source", "source_id": "family", "item_id": "0.jpg"},
|
|
)
|
|
service.process_one()
|
|
op = client.get("/api/v1/operations/" + response.json()["operation_id"]).json()
|
|
assert op["status"] == "FAILED"
|
|
assert len(op["result"]["photo_attempts"]) == 1
|
|
|
|
|
|
def test_random_bad_images_bounded_to_five(client, service, settings, monkeypatch):
|
|
setup_source(service, settings)
|
|
monkeypatch.setattr(
|
|
service.photo_sources,
|
|
"read",
|
|
lambda *args, **kwargs: (b"broken", "bad.jpg", None),
|
|
)
|
|
response = client.post(
|
|
"/api/v1/devices/living-room/upload",
|
|
json={"type": "photo_source", "source_id": "family"},
|
|
)
|
|
service.process_one()
|
|
op = client.get("/api/v1/operations/" + response.json()["operation_id"]).json()
|
|
assert op["status"] == "FAILED"
|
|
assert len(op["result"]["photo_attempts"]) == 5
|
|
row = service.db.one(
|
|
"SELECT write_started FROM operations WHERE id=?", (op["operation_id"],)
|
|
)
|
|
assert row["write_started"] == 0
|
|
|
|
|
|
# Legacy algorithms remain covered; production filesystem access is tested separately.
|
|
pytestmark = pytest.mark.usefixtures("legacy_photo_engine")
|