Celery beat crontab schedules validated and run eagerly: working versions (tested 2026-10-15)

celery==5.6.3
pytest==9.1.1

from celery import Celery
from celery.schedules import crontab

app = Celery("tasks", broker="memory://", backend="cache+memory://")

@app.task
def prune_cache(max_age_hours: int) -> int:
    # real code would delete stale rows and return how many were removed
    return 0

@app.task
def send_report(recipients: list[str]) -> int:
    return len(recipients)

app.conf.beat_schedule = {
    "prune-cache-nightly": {
        "task": "tasks.prune_cache",
        "schedule": crontab(hour=3, minute=0),
        "args": (24,),
    },
    "report-weekdays": {
        "task": "tasks.send_report",
        "schedule": crontab(minute="*/30", hour="9-17", day_of_week="mon-fri"),
        "args": (["[email protected]"],),
    },
}

def validate_beat_schedule(schedule: dict, known_tasks) -> None:
    # beat only fails at runtime in the worker, so check names and types up front
    for name, entry in schedule.items():
        if entry["task"] not in known_tasks:
            raise ValueError(f"{name}: unknown task {entry['task']!r}")
        if not isinstance(entry["schedule"], crontab):
            raise ValueError(f"{name}: schedule must be a crontab")

import pytest
from celery.schedules import crontab
from tasks import app, validate_beat_schedule

@pytest.fixture(autouse=True)
def eager():
    app.conf.update(task_always_eager=True, task_eager_propagates=True)
    yield
    app.conf.update(task_always_eager=False)

def test_shipped_schedule_is_valid():
    validate_beat_schedule(app.conf.beat_schedule, app.tasks)

def test_typo_in_task_name_is_caught():
    bad = {"nightly": {"task": "tasks.prnue_cache", "schedule": crontab()}}
    with pytest.raises(ValueError, match="unknown task"):
        validate_beat_schedule(bad, app.tasks)

def test_crontab_strings_expand_to_sets():
    c = crontab(minute="*/15", hour="9-11")
    assert c.minute == {0, 15, 30, 45}
    assert c.hour == {9, 10, 11}

def test_invalid_crontab_string_fails_fast():
    with pytest.raises(ValueError):
        crontab(minute="bogus")

def test_scheduled_entry_runs_eagerly():
    entry = app.conf.beat_schedule["prune-cache-nightly"]
    assert app.tasks[entry["task"]].apply_async(args=entry["args"]).get() == 0

[pytest]
pythonpath = .