pigway-frame-flow/tests/test_scheduler.py

from datetime import UTC, datetime, timedelta

import pytest

from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import JobInput, PlaylistInput, Schedule
from pigway_frame_flow.scheduler.engine import next_run


@pytest.mark.parametrize("kind,day", [("daily", 4), ("weekdays", 4), ("weekends", 6)])
def test_schedules(kind, day):
    start = datetime(2026, 10, 1, 0, 0, tzinfo=UTC)
    result = datetime.fromisoformat(next_run(Schedule(type=kind, time="09:00"), start))
    assert result.weekday() + 1 == day


def test_cron():
    result = next_run(
        Schedule(type="cron", cron="0 7 * * *"), datetime(2026, 10, 1, 0, 0, tzinfo=UTC)
    )
    assert result.startswith("2026-10-01T23:00")


@pytest.mark.parametrize("type", ["date", "temporary"])
def test_one_time(type):
    current = datetime.now(UTC)
    at = current + timedelta(hours=1)
    assert next_run(Schedule(type=type, at=at), current) == at.isoformat()
    assert next_run(Schedule(type=type, at=at), at) is None


def test_dedupe(app, service):
    scheduler = app.state.scheduler
    job = scheduler.save_job(
        JobInput(
            name="Morning",
            schedule={"type": "daily", "time": "07:00"},
            source={"type": "template", "template": "quote"},
        )
    )
    due = (datetime.now(UTC) - timedelta(seconds=2)).isoformat()
    service.db.execute("UPDATE jobs SET next_run=? WHERE id=?", (due, job["id"]))
    scheduler.tick()
    scheduler.tick()
    assert len(service.db.all("SELECT * FROM operations")) == 1
    assert len(service.db.all("SELECT * FROM job_runs")) == 1


def test_overdue_skipped(app, service):
    scheduler = app.state.scheduler
    job = scheduler.save_job(
        JobInput(
            name="Old",
            schedule={"type": "daily", "time": "07:00"},
            source={"type": "restore"},
        )
    )
    service.db.execute(
        "UPDATE jobs SET next_run=? WHERE id=?",
        ((datetime.now(UTC) - timedelta(hours=1)).isoformat(), job["id"]),
    )
    scheduler.tick()
    assert scheduler.jobs()[0]["last_result"] == "SKIPPED_OVERDUE"
    assert not service.db.all("SELECT * FROM operations")


def test_playlist(app, service):
    scheduler = app.state.scheduler
    scheduler.save_playlist(PlaylistInput(name="Art", image_ids=[1], enabled=True))
    scheduler.tick()
    scheduler.tick()
    assert len(service.db.all("SELECT * FROM operations")) == 1


def test_invalid_cron():
    with pytest.raises(FlowError):
        next_run(Schedule(type="cron", cron="* * * * * *"))


def test_worker_checks_late_sync_before_due_tasks(app, service, monkeypatch):
    scheduler = app.state.scheduler
    calls = []
    monkeypatch.setattr(service, "sync", lambda: calls.append("sync"))

    def tick():
        calls.append("tick")
        scheduler.stop_event.set()

    monkeypatch.setattr(scheduler, "tick", tick)
    monkeypatch.setattr(service, "expire", lambda: None)
    monkeypatch.setattr(service.cloud_library, "sweep", lambda: None)
    monkeypatch.setattr(service.local_retention, "sweep", lambda: None)
    monkeypatch.setattr(service, "process_one", lambda: False)
    scheduler.run()
    assert calls[:2] == ["sync", "tick"]


def test_blocked_job_keeps_retry_slot_without_blocking_worker(
    app, service, monkeypatch
):
    scheduler = app.state.scheduler
    service.registry.drivers["other-room"] = service.registry.get("living-room")
    jobs = []
    for device in ("living-room", "other-room"):
        jobs.append(
            scheduler.save_job(
                JobInput(
                    name=device,
                    device_id=device,
                    schedule={"type": "daily", "time": "07:00"},
                    source={"type": "restore"},
                )
            )
        )
    current = datetime.now(UTC)
    due = (current - timedelta(seconds=2)).isoformat()
    service.db.execute("UPDATE jobs SET next_run=?", (due,))
    unknown = service.submit("display", {}, "living-room", "test", "unknown")
    service.db.execute(
        "UPDATE operations SET status='UNKNOWN',write_started=1 WHERE id=?", (unknown,)
    )
    queued = service.submit("render", {}, "living-room", "test", "render")
    calls = []
    monkeypatch.setattr(service, "sync", lambda: None)
    monkeypatch.setattr(service, "expire", lambda: calls.append("expire"))
    monkeypatch.setattr(service.cloud_library, "sweep", lambda: None)
    monkeypatch.setattr(service.local_retention, "sweep", lambda: None)

    def process():
        calls.append("process")
        assert service.operation(queued)["status"] == "QUEUED"
        scheduler.stop_event.set()
        return False

    monkeypatch.setattr(service, "process_one", process)
    monkeypatch.setattr(
        scheduler.stop_event, "wait", lambda timeout: scheduler.stop_event.set()
    )
    scheduler.run()
    assert calls == ["expire", "process"]
    blocked = service.db.one("SELECT * FROM jobs WHERE id=?", (jobs[0]["id"],))
    assert blocked["next_run"] == due
    assert blocked["last_run"] is None
    assert len(service.db.all("SELECT * FROM job_runs")) == 1
    assert service.db.one("SELECT * FROM job_runs")["job_id"] == jobs[1]["id"]
    scheduler.tick(current + timedelta(minutes=6))
    assert (
        service.db.one("SELECT * FROM jobs WHERE id=?", (jobs[0]["id"],))["last_result"]
        == "SKIPPED_OVERDUE"
    )