Testing Scheduled Tasks in Python/Flask

Question

I'm building a small Flask app the runs a job every so often, say once a month. I'm using apschduler for this. Is there a way to set up a test for this? I'm using py.test.

Something like:

def test_cron(self, app):
    wait(months=1)
    assert cron_has_run

Any suggestions?

Berlinda

Answer 1

You shouldn't need to test that the job actually runs, since that is handled by apscheduler, which is already tested. Instead, test the task itself and test that you scheduled it as expected.

You can inspect the scheduled jobs using get_jobs. If you want to test that you scheduled a job correctly, schedule it then find its job in the list. You can test if next_run_time is a month out, for example.

for job in get_jobs():
    if name != 'name of job you care about':
        continue

    assert job.next_run_time >= datetime.utcnow() + timedelta(days=30)
    assert job.next_run_time <= datetime.utcnow() + timedelta(days=32)

If you're interested in how apscheduler tests its scheduler, you can look at the code: https://github.com/agronholm/apscheduler/blob/master/tests/test_schedulers.py. It's using mocks and dates in the past.

Chalutz

Answer 2

If you're using Flask-APScheduler and pytest, you can do something like:

import datetime

from apscheduler.job import Job
from apscheduler.triggers.cron import (
    BaseField,
    CronTrigger,
    DayOfMonthField,
    DayOfWeekField,
    MonthField,
    WeekField,
)

from app.core.extensions import scheduler

from ..tasks import a_scheduled_function


def test_is_scheduled_hourly():
    """
    Test to verify that `a_scheduled_function` is scheduled correctly on the 2nd minute
    of every hour using a cron trigger.
    """
    # Retrieve scheduled job using its id
    a_scheduled_function_job: Job = scheduler.get_job(id="a_scheduled_function")

    assert a_scheduled_function_job is not None
    assert a_scheduled_function_job.func == a_scheduled_function

    trigger: CronTrigger = a_scheduled_function_job.trigger

    assert isinstance(trigger, CronTrigger)
    assert trigger.timezone == datetime.timezone.utc
    assert trigger.end_date is None
    assert trigger.fields == [
        BaseField("year", "*"),
        MonthField("month", "*"),
        DayOfMonthField("day", "*"),
        WeekField("week", "*"),
        DayOfWeekField("day_of_week", "*"),
        BaseField("hour", "*"),
        BaseField("minute", "2"),
        BaseField("second", "0"),
    ]

This is a sample test for a function that is scheduled to run every hour on the second minute.

Pirali