pigway-frame-flow/tests/test_recovery.py

from datetime import UTC, datetime, timedelta

import pytest
from test_display import submit

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


def unknown_upload(service, monkeypatch):
    def fail(*args):
        raise FlowError(ErrorCode.WRITE_RESULT_UNKNOWN, "Response lost", 502)

    monkeypatch.setattr(service.registry.get("living-room"), "upload", fail)
    operation = submit(service)
    service.process_one()
    return operation


def test_missing_upload_is_visible_without_unsafe_unlock(client, service, monkeypatch):
    operation = unknown_upload(service, monkeypatch)
    service.sync()
    recovery = client.get("/api/v1/operations/" + operation).json()["recovery"]
    assert recovery["reason"] == "gallery_match_missing"
    assert recovery["state"] == "checking"
    assert not recovery["requires_manual_review"]
    assert service.blocked("living-room")
    assert service.registry.get("living-room").client.uploads == 0


def test_persistent_block_escalates_once_survives_restart_and_preserves_evidence(
    service, monkeypatch
):
    operation = unknown_upload(service, monkeypatch)
    original = service.operation(operation)
    service.db.execute(
        "UPDATE operations SET created_at=? WHERE id=?",
        (
            (datetime.now(UTC) - timedelta(hours=1)).isoformat(),
            operation,
        ),
    )
    service.sync()
    first = service.operation(operation)["recovery"]
    assert first["state"] == "manual_review"
    assert first["reason"] == "gallery_match_missing"
    service.unknown_checked.clear()
    service.sync()
    restarted = DisplayService(
        service.db, service.settings, service.registry, service.assets, service.renderer
    )
    restarted.sync()
    assert (
        len(service.db.all("SELECT * FROM events WHERE event_type='operation.blocked'"))
        == 1
    )
    assert (
        service.db.one(
            "SELECT COUNT(*) AS n FROM history WHERE result='MANUAL_REVIEW_REQUIRED'"
        )["n"]
        == 1
    )
    assert service.operation(operation)["recovery"]["checks"] == 3
    current_result = service.operation(operation)["result"]
    assert current_result.pop("sync_observations")
    assert current_result == original["result"]
    service.reconcile(operation, acknowledge=True)
    assert service.operation(operation)["recovery"]["state"] == "resolved"
    assert not service.operation(operation)["recovery"]["requires_manual_review"]
    assert not service.blocked("living-room")


def test_missing_legacy_evidence_escalates_without_cloud_write(service):
    operation = submit(service)
    service._start_write(operation)
    service.update(operation, "UNKNOWN")
    service.sync()
    assert (
        service.operation(operation)["recovery"]["reason"] == "missing_upload_evidence"
    )
    assert service.blocked("living-room")


def test_delete_unknown_requires_manual_review_without_replay(service):
    operation = service.submit(
        "delete", {"image_id": 1}, "living-room", "admin", "test"
    )
    service._start_write(operation)
    service.update(operation, "UNKNOWN")
    service.sync()
    assert (
        service.operation(operation)["recovery"]["reason"]
        == "manual_reconciliation_required"
    )
    assert service.blocked("living-room")
    assert service.registry.get("living-room").images()[0].image_id == 1


def test_malformed_reconciliation_does_not_abort_other_background_work(
    service, monkeypatch
):
    operation = unknown_upload(service, monkeypatch)

    def invalid_metadata(*args):
        raise ValueError("vendor metadata with private data")

    monkeypatch.setattr(service.upload_reconciliation, "check", invalid_metadata)
    service.sync()
    op = service.operation(operation)
    assert op["recovery"]["reason"] == "reconciliation_error"
    assert "private" not in str(op["recovery"])
    assert service.blocked("living-room")


@pytest.mark.parametrize("status", ["SYNCING", "UNKNOWN"])
def test_later_confirmed_display_closes_earlier_syncing_attempt_without_replay(
    service, status
):
    first = submit(service)
    service.process_one()
    assert service.operation(first)["status"] == "SYNCING"
    second = submit(service)
    service.process_one()
    driver = service.registry.get("living-room")
    uploads = driver.client.uploads
    previous_error = {"code": "WRITE_RESULT_UNKNOWN", "message": "Refresh unconfirmed"}
    if status == "UNKNOWN":
        service.update(first, status, error=previous_error)
    service.sync()
    assert service.operation(second)["status"] == "COMPLETED"
    old = service.operation(first)
    assert old["status"] == "FAILED"
    assert old["error"]["code"] == "DISPLAY_NOT_CONFIRMED"
    assert old["error"]["details"]["superseded_by_operation_id"] == second
    if status == "UNKNOWN":
        assert old["error"]["details"]["previous_error"] == previous_error
    assert driver.client.uploads == uploads
    assert not service.blocked("living-room")
    service.sync()
    assert (
        len(
            service.db.all(
                "SELECT * FROM history WHERE operation_id=? AND result='DISPLAY_NOT_CONFIRMED'",
                (first,),
            )
        )
        == 1
    )


def test_unconfirmed_later_display_does_not_close_old_attempt(service):
    first = submit(service)
    service.process_one()
    second = submit(service)
    service.process_one()
    service.recovery.close_superseded_displays()
    assert service.operation(first)["status"] == "SYNCING"
    assert service.operation(second)["status"] == "SYNCING"


def test_confirmed_temporary_override_does_not_close_earlier_display(service):
    first = submit(service)
    service.process_one()
    second = submit(service, 60, True)
    service.process_one()
    service.sync()
    assert service.operation(second)["status"] == "COMPLETED"
    assert service.operation(first)["status"] == "SYNCING"


@pytest.mark.parametrize("client_id", ["scheduler", "scheduler-retry"])
def test_automatic_upload_waits_for_prior_sync_confirmation(service, client_id):
    first = submit(service)
    service.process_one()
    payload = service.db.one("SELECT payload FROM operations WHERE id=?", (first,))[
        "payload"
    ]
    import json

    second = service.submit(
        "display", json.loads(payload), "living-room", client_id, "fixture"
    )
    uploads = service.registry.get("living-room").client.uploads
    service.process_one()
    assert service.operation(second)["status"] == "QUEUED"
    assert service.registry.get("living-room").client.uploads == uploads
    service.sync()
    assert service.operation(first)["status"] == "COMPLETED"
    service.process_one()
    service.sync()
    assert service.operation(second)["status"] == "COMPLETED"
    assert service.registry.get("living-room").client.uploads == uploads + 1


def test_device_evidence_is_bounded_and_does_not_extend_sync_timeout(service):
    first = submit(service)
    service.process_one()
    op = service.db.one("SELECT * FROM operations WHERE id=?", (first,))
    before = service.operation(first)["updated_at"]
    device = service.registry.get("living-room").device()
    for i in range(20):
        sample = device.model_copy(
            update={
                "current_content": device.current_content.model_copy(
                    update={"id": 1000 + i}
                )
            }
        )
        service.recovery.observe_device(op, device=sample)
    result = service.operation(first)
    samples = result["result"]["sync_observations"]
    assert len(samples) == 12
    assert samples[0]["image_id"] == 1000 and samples[-1]["image_id"] == 1019
    assert result["updated_at"] == before
    service.update(first, "UNKNOWN", {"image_id": 7})
    assert service.operation(first)["result"]["sync_observations"] == samples


def test_device_read_failure_evidence_never_copies_raw_error_details(service):
    first = submit(service)
    service.process_one()
    op = service.db.one("SELECT * FROM operations WHERE id=?", (first,))
    service.recovery.observe_device(
        op,
        error=FlowError(
            ErrorCode.CLOUD_UNAVAILABLE,
            "unavailable",
            details={
                "transport_error": "read_timeout",
                "url": "private-url-must-not-be-stored",
            },
        ),
    )
    sample = service.operation(first)["result"]["sync_observations"][-1]
    assert sample["read_error"]["transport_error"] == "read_timeout"
    assert "private-url" not in str(sample)


def test_sync_timeout_includes_observed_device_state(service):
    first = submit(service)
    service.process_one()
    service.registry.get("living-room").client.synced = False
    service.registry.get("living-room").invalidate()
    service.settings.sync_timeout = 0
    service.sync()
    op = service.operation(first)
    assert op["status"] == "UNKNOWN"
    assert op["error"]["details"]["last_device_observation"]["synced"] is False


def test_automatic_wait_does_not_extend_past_sync_timeout(service):
    import json

    first = submit(service)
    service.process_one()
    payload = service.db.one("SELECT payload FROM operations WHERE id=?", (first,))[
        "payload"
    ]
    service.db.execute(
        "UPDATE operations SET updated_at=? WHERE id=?",
        (
            (
                datetime.now(UTC) - timedelta(seconds=service.settings.sync_timeout + 1)
            ).isoformat(),
            first,
        ),
    )
    second = service.submit(
        "display", json.loads(payload), "living-room", "scheduler", "fixture"
    )
    service.process_one()
    assert service.operation(second)["status"] == "SYNCING"