78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
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="* * * * * *"))
|