pigway-frame-flow/tests/test_retention.py

from datetime import UTC, datetime, timedelta

import pytest
from PIL import Image

from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import JobInput, Schedule
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
from pigway_frame_flow.storage.database import encode


def upload_random(client, 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=".")
    )
    r = client.post(
        "/api/v1/display",
        json={
            "device_id": "living-room",
            "content": {"type": "photo_source", "source_id": "family"},
            "options": {"cleanup_unstarred": True},
        },
    )
    assert r.status_code == 202
    service.process_one()
    op = client.get("/api/v1/operations/" + r.json()["operation_id"]).json()
    return op["result"]["image_id"]


def due(service, image_id):
    service.db.execute(
        "UPDATE image_retention SET expires_at=? WHERE image_id=?",
        ((datetime.now(UTC) - timedelta(days=1)).isoformat(), image_id),
    )
    # The test driver switches to a different image, not a real device.
    driver = service.registry.get("living-room")
    driver.display(1)
    service.db.execute(
        "UPDATE device_state SET previous_image_id=NULL,base_image_id=NULL WHERE device_id=?",
        ("living-room",),
    )


def test_register_new_random_upload_with_seven_day_expiry(client, service, settings):
    image_id = upload_random(client, service, settings)
    row = service.db.one("SELECT * FROM image_retention WHERE image_id=?", (image_id,))
    delta = datetime.fromisoformat(row["expires_at"]) - datetime.now(UTC)
    assert 6.99 < delta.total_seconds() / 86400 <= 7
    assert row["upload_operation"] and not row["starred"]


def test_star_api_and_gallery_persist(client, service, settings):
    image_id = upload_random(client, service, settings)
    r = client.put(
        f"/api/v1/devices/living-room/images/{image_id}/star", json={"starred": True}
    )
    assert r.status_code == 200 and r.json()["starred"]
    items = client.get("/api/v1/devices/living-room/images").json()
    assert next(x for x in items if x["image_id"] == image_id)["starred"]
    due(service, image_id)
    service.retention.sweep()
    assert (
        service.db.one(
            "SELECT cleanup_operation FROM image_retention WHERE image_id=?",
            (image_id,),
        )["cleanup_operation"]
        is None
    )


def test_due_cleanup_deletes_only_owned_upload(client, service, settings):
    image_id = upload_random(client, service, settings)
    due(service, image_id)
    service.retention.sweep()
    row = service.db.one("SELECT * FROM image_retention WHERE image_id=?", (image_id,))
    assert row["cleanup_operation"]
    service.process_one()
    assert (
        service.db.one("SELECT * FROM image_retention WHERE image_id=?", (image_id,))
        is None
    )
    assert 1 in {x.image_id for x in service.registry.get("living-room").images()}


def test_current_and_restore_and_playlist_protected(client, service, settings):
    image_id = upload_random(client, service, settings)
    service.db.execute(
        "UPDATE image_retention SET expires_at=? WHERE image_id=?",
        ("2000-01-01T00:00:00+00:00", image_id),
    )
    service.retention.sweep()
    assert not service.db.one(
        "SELECT cleanup_operation FROM image_retention WHERE image_id=?", (image_id,)
    )["cleanup_operation"]
    due(service, image_id)
    service.db.execute("UPDATE device_state SET previous_image_id=?", (image_id,))
    assert not service.retention.eligible("living-room", image_id)
    service.db.execute("UPDATE device_state SET previous_image_id=NULL")
    service.db.execute(
        "INSERT INTO playlists VALUES(?,?,0,NULL)",
        ("p", encode({"device_id": "living-room", "image_ids": [image_id]})),
    )
    assert not service.retention.eligible("living-room", image_id)


def test_star_between_queue_and_execution_cancels_delete(client, service, settings):
    image_id = upload_random(client, service, settings)
    due(service, image_id)
    service.retention.sweep()
    assert (
        client.put(
            f"/api/v1/devices/living-room/images/{image_id}/star",
            json={"starred": True},
        ).status_code
        == 200
    )
    service.process_one()
    assert image_id in {
        x.image_id for x in service.registry.get("living-room").images()
    }


def test_unknown_delete_is_not_retried(client, service, settings, monkeypatch):
    image_id = upload_random(client, service, settings)
    due(service, image_id)
    calls = []

    def unknown(id):
        calls.append(id)
        raise FlowError(E.WRITE_RESULT_UNKNOWN, "Unknown delete result")

    monkeypatch.setattr(service.registry.get("living-room"), "delete", unknown)
    service.retention.sweep()
    service.process_one()
    service.retention.checked = 0
    service.retention.sweep()
    assert calls == [image_id]
    assert service.blocked("living-room")


def test_cleanup_option_requires_random_photo_job():
    with pytest.raises(ValueError):
        JobInput(
            name="x",
            schedule=Schedule(type="daily", time="07:00"),
            source={"type": "template", "template": "quote"},
            cleanup_unstarred=True,
        )


def test_first_cleanup_runs_after_recent_host_boot(
    client, service, settings, monkeypatch
):
    image_id = upload_random(client, service, settings)
    due(service, image_id)
    monkeypatch.setattr(
        "pigway_frame_flow.display.retention.time.monotonic", lambda: 120
    )
    service.retention.sweep()
    row = service.db.one("SELECT * FROM image_retention WHERE image_id=?", (image_id,))
    assert row["cleanup_operation"]
    assert service.retention.checked == 120
    service.retention.sweep()
    assert (
        service.db.one("SELECT COUNT(*) AS n FROM operations WHERE action='delete'")[
            "n"
        ]
        == 1
    )


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