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_no_retry_unknown(service):
    driver = service.registry.get("living-room")

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

    driver.upload = timeout
    id = submit(service)
    service.process_one()
    assert service.operation(id)["status"] == "UNKNOWN"
    with pytest.raises(FlowError):
        submit(service)
    assert service.process_one() is False
    assert service.reconcile(id)["resolved"] is False
    assert service.reconcile(id, True)["resolved"] is True
    assert service.operation(id)["status"] == "FAILED"


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"


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"