pigway-frame-flow/tests/test_local_cloud_library.py

from io import BytesIO

import pytest
from PIL import Image

from pigway_frame_flow.providers.photo_sources import PhotoSourceInput


def setup_photos(service, settings):
    root = settings.data_dir.parent / "photos"
    root.mkdir()
    settings.photo_root = root
    Image.new("RGB", (300, 200)).save(root / "a.jpg")
    service.photo_sources.save(
        "family", PhotoSourceInput(name="Family", kind="local", location=".")
    )


def make_job(client, name):
    r = client.post(
        "/api/v1/jobs",
        json={
            "name": name,
            "enabled": False,
            "schedule": {"type": "daily", "time": "07:00"},
            "source": {"type": "photo_source", "source_id": "family"},
        },
    )
    assert r.status_code == 201
    return r.json()["id"]


def run(client, service, job, confirm=True):
    r = client.post(f"/api/v1/jobs/{job}/run")
    assert r.status_code == 202, r.text
    id = r.json()["operation_id"]
    service.process_one()
    if confirm:
        service.sync()
    op = client.get("/api/v1/operations/" + id).json()
    return op["result"]["image_id"], op["result"]["rendered_asset_id"]


def test_task_origin_filter_and_star_bulk(client, service, settings):
    setup_photos(service, settings)
    job = make_job(client, "Family random")
    image, asset = run(client, service, job)
    rows = client.get("/api/v1/assets", params={"origin": "task", "job_id": job}).json()
    assert len(rows) == 1 and rows[0]["job_name"] == "Family random"
    r = client.post(
        "/api/v1/assets/batch", json={"action": "star", "asset_ids": [asset, asset]}
    )
    assert len(r.json()["results"]) == 1
    assert client.get("/api/v1/assets").json()[0]["starred"]
    r = client.post(
        "/api/v1/assets/batch",
        json={"action": "delete", "asset_ids": [asset, "missing"]},
    )
    assert [x["status"] for x in r.json()["results"]] == ["completed", "failed"]


def test_each_task_keeps_one_after_confirmation(client, service, settings):
    setup_photos(service, settings)
    job = make_job(client, "A")
    first, _ = run(client, service, job)
    second, _ = run(client, service, job)
    service.cloud_library.sweep()
    assert service.process_one()
    assert first not in {
        x.image_id for x in service.registry.get("living-room").images()
    }
    assert second in {x.image_id for x in service.registry.get("living-room").images()}
    other = make_job(client, "B")
    third, _ = run(client, service, other)
    service.cloud_library.checked = 0
    service.cloud_library.sweep()
    assert second in {x.image_id for x in service.registry.get("living-room").images()}
    assert third in {x.image_id for x in service.registry.get("living-room").images()}


def test_unconfirmed_replacement_never_evicts_previous(client, service, settings):
    setup_photos(service, settings)
    job = make_job(client, "A")
    first, _ = run(client, service, job)
    second, _ = run(client, service, job, confirm=False)
    service.cloud_library.sweep()
    assert not service.process_one()
    assert first in {x.image_id for x in service.registry.get("living-room").images()}
    assert not service.db.one(
        "SELECT cleanup_operation FROM managed_cloud_uploads WHERE image_id=?", (first,)
    )["cleanup_operation"]


def test_starred_old_upload_survives_then_unstar_cleans(client, service, settings):
    setup_photos(service, settings)
    job = make_job(client, "A")
    first, asset = run(client, service, job)
    assert (
        client.put(f"/api/v1/assets/{asset}/star", json={"starred": True}).status_code
        == 200
    )
    run(client, service, job)
    service.cloud_library.sweep()
    assert not service.process_one()
    assert (
        client.put(f"/api/v1/assets/{asset}/star", json={"starred": False}).status_code
        == 200
    )
    service.cloud_library.checked = 0
    service.cloud_library.sweep()
    assert service.process_one()
    assert first not in {
        x.image_id for x in service.registry.get("living-room").images()
    }
    assert service.assets.path(asset).exists()


def test_manual_upload_origin(client, service):
    buf = BytesIO()
    Image.new("RGB", (30, 20)).save(buf, "JPEG")
    r = client.post(
        "/api/v1/preview", files={"file": ("manual.jpg", buf.getvalue(), "image/jpeg")}
    )
    id = r.json()["asset_id"]
    r = client.post(
        "/api/v1/devices/living-room/upload", json={"type": "image", "asset_id": id}
    )
    service.process_one()
    rows = client.get("/api/v1/assets", params={"origin": "manual"}).json()
    assert len(rows) == 1 and rows[0]["asset_id"] == id


def test_clear_cloud_requires_confirmation_and_clears_current(client, service):
    before = len(service.registry.get("living-room").images())
    assert (
        client.post(
            "/api/v1/devices/living-room/images/clear", json={"confirm_current": False}
        ).status_code
        == 400
    )
    r = client.post(
        "/api/v1/devices/living-room/images/clear", json={"confirm_current": True}
    )
    assert r.status_code == 200 and len(r.json()["operations"]) == before
    for _ in range(before):
        assert service.process_one()
    assert not service.registry.get("living-room").images()


def test_manual_display_does_not_replace_task_head(client, service, settings):
    setup_photos(service, settings)
    job = make_job(client, "A")
    first, asset = run(client, service, job)
    second, _ = run(client, service, job)
    client.put(f"/api/v1/assets/{asset}/star", json={"starred": True})
    response = client.post(
        "/api/v1/display", json={"content": {"type": "image", "asset_id": asset}}
    )
    service.process_one()
    service.sync()
    op = client.get("/api/v1/operations/" + response.json()["operation_id"]).json()
    row = service.db.one(
        "SELECT job_id FROM managed_cloud_uploads WHERE image_id=?",
        (op["result"]["image_id"],),
    )
    assert row["job_id"] is None
    assert service.cloud_library.retained(
        service.db.one(
            "SELECT * FROM managed_cloud_uploads WHERE image_id=?", (second,)
        )
    )


# Legacy algorithms remain covered; production filesystem access is tested separately.
pytestmark = pytest.mark.usefixtures("legacy_photo_engine")