155 lines
4.6 KiB
Python
155 lines
4.6 KiB
Python
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"
|