pigway-frame-flow/tests/test_photo_preferences.py

import json
from io import BytesIO

import pytest
from PIL import Image

from pigway_frame_flow.core.photo_preferences import eligible, flags, set_flags
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import AutoCollageContent, PhotoSourceContent
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput


def photo(service, name, color):
    data = BytesIO()
    Image.new("RGB", (320, 240), color).save(data, "JPEG")
    return service.assets.put(data.getvalue(), name)


def test_local_exclusion_star_and_restore(client, service):
    excluded = photo(service, "excluded.jpg", "red")["asset_id"]
    allowed = photo(service, "allowed.jpg", "blue")["asset_id"]
    assert (
        client.put(
            f"/api/v1/assets/{excluded}/star", json={"starred": True}
        ).status_code
        == 200
    )
    result = client.put(f"/api/v1/assets/{excluded}/exclusion", json={"excluded": True})
    assert (
        result.status_code == 200
        and result.json()["excluded"]
        and result.json()["starred"]
    )
    assert next(
        x for x in client.get("/api/v1/assets").json() if x["asset_id"] == excluded
    )["excluded"]
    asset = service.content_asset(
        AutoCollageContent(type="auto_collage", count=1, use_ai=False)
    )
    saved = json.loads(
        service.db.one(
            "SELECT payload FROM asset_sources WHERE asset_id=?", (asset["asset_id"],)
        )["payload"]
    )
    assert [r["asset_id"] for r in saved["data"]["layout"]["regions"]] == [allowed]
    assert service.assets.path(excluded).is_file()
    assert (
        client.post(
            "/api/v1/assets/batch", json={"asset_ids": [allowed], "action": "exclude"}
        ).json()["results"][0]["status"]
        == "completed"
    )
    with pytest.raises(FlowError):
        service.content_asset(
            AutoCollageContent(type="auto_collage", count=1, use_ai=False)
        )
    restored = client.put(
        f"/api/v1/assets/{excluded}/exclusion", json={"excluded": False}
    ).json()
    assert not restored["excluded"] and restored["starred"]
    assert service.content_asset(
        AutoCollageContent(type="auto_collage", count=1, use_ai=False)
    )


def test_source_preferences_are_scoped_and_do_not_mutate_cached_catalog(
    client, service, monkeypatch
):
    source = PhotoSourceInput(
        name="Drive", kind="webdav", location="https://photos.example.test/dav"
    )
    items = [{"id": "folder/photo.jpg", "name": "photo.jpg"}]
    monkeypatch.setattr(service.photo_sources, "get", lambda _: source)
    monkeypatch.setattr(service.photo_sources, "catalog", lambda *args, **kwargs: items)
    url = "/api/v1/photo-sources/drive/preferences"
    assert (
        client.put(url, json={"item_id": items[0]["id"], "starred": True}).status_code
        == 200
    )
    assert client.put(
        url, json={"item_id": items[0]["id"], "excluded": True}
    ).json() == {"starred": True, "excluded": True}
    result = client.get("/api/v1/photo-sources/drive/images").json()["items"][0]
    assert result["starred"] and result["excluded"]
    assert "excluded" not in items[0]
    assert eligible(service.db, "drive", items) == []
    assert eligible(service.db, "another-drive", items) == items
    assert (
        client.put(url, json={"item_id": "missing", "excluded": True}).status_code
        == 404
    )
    with pytest.raises(FlowError):
        service.content_asset(
            AutoCollageContent(
                type="auto_collage", source_id="drive", count=1, use_ai=False
            )
        )
    with pytest.raises(FlowError):
        service.photo_sources.read("drive")
    with pytest.raises(FlowError):
        service.content_asset(
            PhotoSourceContent(type="photo_source", source_id="drive")
        )
    client.put(url, json={"item_id": items[0]["id"], "excluded": False})
    assert eligible(service.db, "drive", items) == items
    assert flags(service.db, "drive", items[0]["id"]) == {
        "starred": True,
        "excluded": False,
    }


def test_preference_persistence(settings):
    from pigway_frame_flow.storage.database import Database

    path = settings.data_dir / "preferences.db"
    path.parent.mkdir(parents=True, exist_ok=True)
    db = Database(path)
    set_flags(db, "drive", "item", excluded=True)
    db.conn.close()
    reopened = Database(path)
    assert flags(reopened, "drive", "item")["excluded"]
    reopened.conn.close()