pigway-frame-flow/tests/test_display.py

from datetime import UTC, datetime, timedelta

import pytest

from pigway_frame_flow.display.service import DisplayService
from pigway_frame_flow.errors import ErrorCode, FlowError
from pigway_frame_flow.models import DisplayRequest


def submit(service, duration=0, restore=False, priority=50, key=None, content=None):
    request = DisplayRequest(
        content=content or {"type": "template", "template": "quote", "data": {}},
        options={
            "duration": duration,
            "restore_previous": restore,
            "priority": priority,
        },
    )
    return service.submit(
        "display",
        request.model_dump(mode="json"),
        "living-room",
        "admin",
        "request-test",
        key,
    )


def expired(service, id):
    service.db.execute(
        "UPDATE overrides SET expires_at=? WHERE id=?",
        ((datetime.now(UTC) - timedelta(seconds=1)).isoformat(), id),
    )


def test_display_restore(service):
    id = submit(service, 60, True)
    service.process_one()
    service.sync()
    assert service.operation(id)["status"] == "COMPLETED"
    assert service.registry.get("living-room").get_current_image() == 2
    expired(service, id)
    service.expire()
    service.sync()
    assert service.registry.get("living-room").get_current_image() == 1


def test_priority_lower_defers_upload(service):
    high = submit(service, 300, True, 90)
    service.process_one()
    low = submit(service, 60, True, 10)
    service.process_one()
    assert service.operation(low)["status"] == "QUEUED"
    assert service.registry.get("living-room").client.uploads == 1
    expired(service, high)
    service.expire()
    service.process_one()
    assert service.registry.get("living-room").client.uploads == 2


def test_nested_overrides(service):
    a = submit(service, 300, True, 10)
    service.process_one()
    b = submit(service, 60, True, 90)
    service.process_one()
    expired(service, b)
    service.expire()
    assert service.registry.get("living-room").get_current_image() == 2
    expired(service, a)
    service.expire()
    assert service.registry.get("living-room").get_current_image() == 1


def test_unknown_does_not_block_new_independent_upload(service):
    driver = service.registry.get("living-room")
    attempts = []

    def timeout(*args):
        attempts.append(args)
        raise FlowError(ErrorCode.WRITE_RESULT_UNKNOWN, "Write outcome unknown", 502)

    driver.upload = timeout
    first = submit(service)
    service.process_one()
    assert service.operation(first)["status"] == "UNKNOWN"
    second = submit(service, priority=10)
    assert second != first
    assert service.process_one()
    assert service.operation(first)["status"] == "FAILED"
    assert service.operation(second)["status"] == "UNKNOWN"
    assert len(attempts) == 2


def test_restart_preserves_override(service):
    id = submit(service, 60, True)
    service.process_one()
    restarted = DisplayService(
        service.db, service.settings, service.registry, service.assets, service.renderer
    )
    expired(restarted, id)
    restarted.expire()
    assert service.registry.get("living-room").get_current_image() == 1


def test_restart_marks_inflight_unknown(service):
    id = submit(service)
    service._start_write(id)
    DisplayService(
        service.db, service.settings, service.registry, service.assets, service.renderer
    )
    assert service.operation(id)["status"] == "UNKNOWN"


def test_idempotency(service):
    first = submit(service, key="repeat")
    second = submit(service, key="repeat")
    assert first == second
    service.process_one()
    service.process_one()
    assert service.registry.get("living-room").client.uploads == 1
    with pytest.raises(FlowError):
        submit(service, priority=10, key="repeat")


def test_manual_change_prevents_restore(service):
    id = submit(service, 60, True)
    service.process_one()
    driver = service.registry.get("living-room")
    driver.client.display(1)
    driver.invalidate()
    expired(service, id)
    service.expire()
    assert driver.get_current_image() == 1
    assert not service.db.all("SELECT * FROM overrides")


def test_delete_restore_reference_protected(service):
    submit(service, 60, True)
    service.process_one()
    delete = service.submit(
        "delete",
        {"image_id": 1, "confirm_current": True},
        "living-room",
        "admin",
        "delete-test",
    )
    service.process_one()
    assert service.operation(delete)["status"] == "FAILED"
    assert service.operation(delete)["error"]["code"] == "CONFLICT"
    assert (
        service.operation(delete)["error"]["details"]["reason"]
        == "RESTORE_IMAGE_PROTECTED"
    )


@pytest.mark.parametrize("automatic", [True, False])
def test_completed_restore_consumes_self_reference(client, service, automatic):
    display = submit(service, 60, True)
    service.process_one()
    service.sync()
    if automatic:
        expired(service, display)
        service.expire()
    else:
        service.submit("restore", {}, "living-room", "admin", "restore-test")
        service.process_one()
    service.sync()
    assert service.state("living-room")["previous_image_id"] is None
    assert not client.get("/api/v1/devices/living-room").json()["can_restore"]
    delete = service.submit(
        "delete",
        {"image_id": 1, "confirm_current": True},
        "living-room",
        "admin",
        "delete-restored",
    )
    service.process_one()
    assert service.operation(delete)["status"] == "COMPLETED"


@pytest.mark.parametrize("via_delete", [False, True])
def test_existing_self_reference_is_repaired(client, service, via_delete):
    service.state("living-room")
    service.db.execute(
        "UPDATE device_state SET previous_image_id=1,managed_image_id=1 WHERE device_id=?",
        ("living-room",),
    )
    if via_delete:
        operation = service.submit(
            "delete",
            {"image_id": 1, "confirm_current": True},
            "living-room",
            "admin",
            "delete-stale-reference",
        )
        service.process_one()
        assert service.operation(operation)["status"] == "COMPLETED"
    else:
        assert not client.get("/api/v1/devices/living-room?refresh=true").json()[
            "can_restore"
        ]
    assert service.state("living-room")["previous_image_id"] is None


@pytest.mark.parametrize(
    "protection",
    ["override", "base", "unsynced", "unknown", "syncing", "other_previous"],
)
def test_self_reference_cleanup_preserves_real_restore_requirements(
    service, protection
):
    service.state("living-room")
    service.db.execute(
        "UPDATE device_state SET previous_image_id=1,managed_image_id=1 WHERE device_id=?",
        ("living-room",),
    )
    device = service.registry.get("living-room").device()
    if protection == "override":
        service.db.execute(
            "INSERT INTO overrides VALUES(?,?,?,?,?,?)",
            ("active", "living-room", 1, 50, "2099-01-01", "2026-01-01"),
        )
    elif protection == "base":
        service.db.execute(
            "UPDATE device_state SET base_image_id=1 WHERE device_id=?",
            ("living-room",),
        )
    elif protection == "unsynced":
        device = device.model_copy(
            update={
                "current_content": device.current_content.model_copy(
                    update={"synced": False}
                )
            }
        )
    elif protection in ("unknown", "syncing"):
        operation = submit(service)
        service._start_write(operation)
        service.update(operation, protection.upper())
    else:
        service.db.execute(
            "UPDATE device_state SET previous_image_id=2 WHERE device_id=?",
            ("living-room",),
        )
    service.clear_self_restore_reference("living-room", device)
    assert service.state("living-room")["previous_image_id"] == (
        2 if protection == "other_previous" else 1
    )


def test_offline(service):
    service.registry.get("living-room").client.online = False
    id = submit(service)
    service.process_one()
    assert service.operation(id)["error"]["code"] == "DEVICE_OFFLINE"


def test_late_sync_resolves_unknown_without_upload_retry(service, monkeypatch):
    id = submit(service)
    service.process_one()
    service.update(id, "UNKNOWN")
    driver = service.registry.get("living-room")
    calls = []
    device = driver.device
    monkeypatch.setattr(driver, "device", lambda **kw: calls.append(kw) or device(**kw))
    service.sync()
    assert service.operation(id)["status"] == "COMPLETED"
    assert driver.client.uploads == 1
    assert calls == [{"refresh": True}]
    assert not service.blocked("living-room")
    assert service.db.one(
        "SELECT confirmed_at FROM managed_cloud_uploads WHERE operation_id=?", (id,)
    )["confirmed_at"]


def test_unknown_wrong_target_stays_locked_and_queries_at_most_once_per_minute(
    service, monkeypatch
):
    id = submit(service)
    service.process_one()
    service.update(id, "UNKNOWN")
    driver = service.registry.get("living-room")
    driver.client.display(1)
    driver.invalidate()
    calls = []
    device = driver.device
    monkeypatch.setattr(driver, "device", lambda **kw: calls.append(kw) or device(**kw))
    service.sync()
    service.sync()
    assert service.operation(id)["status"] == "UNKNOWN"
    assert service.blocked("living-room")
    assert driver.client.uploads == 1
    assert len(calls) == 1


def test_unknown_matching_image_without_sync_stays_locked(service, monkeypatch):
    id = submit(service)
    service.process_one()
    service.update(id, "UNKNOWN")
    driver = service.registry.get("living-room")
    device = driver.device()
    device.current_content.synced = False
    monkeypatch.setattr(driver, "device", lambda **kw: device)
    service.sync()
    assert service.operation(id)["status"] == "UNKNOWN"
    assert service.blocked("living-room")
    assert driver.client.uploads == 1