Testing schedules#

In this guide, we'll show you how to use the Dagster UI and Python to test your schedules. For more information about testing in Dagster, refer to the Testing documentation.

Testing schedules in the Dagster UI#

Using the UI, you can manually trigger test evaluations of a schedule and view the results. This can be helpful when creating a schedule or for troubleshooting unexpected scheduling behavior.

1. In the UI, click Overview > Schedules tab.

2. Click the schedule you want to test.

3. Click the Test Schedule button, located near the top right corner of the page.

4. You'll be prompted to select a mock schedule evaluation time. As schedules are defined on a cadence, the evaluation times in the dropdown are past and future times along that cadence.

   For example, let's say you're testing a schedule with a cadence of "Every day at X time". In the dropdown, you'd see past and future evaluation times along that cadence:

   

5. After selecting an evaluation time, click the Evaluate button.

A window containing the evaluation result will display after the test completes. If the evaluation was successful, click Open in Launchpad to launch a run with the same config as the test evaluation.

Testing schedules in Python#

You can also test your schedules directly in Python. In this section, we'll demonstrate how to test:

• @schedule-decorated functions
• Schedules with resources

Testing @schedule-decorated functions#

To test a function decorated by the @schedule decorator, you can invoke the schedule definition like it's a regular Python function. The invocation will return run config, which can then be validated using the validate_run_config function.

Let's say we want to test the configurable_job_schedule in this example:

@op(config_schema={"scheduled_date": str})
def configurable_op(context: OpExecutionContext):
    context.log.info(context.op_config["scheduled_date"])


@job
def configurable_job():
    configurable_op()


@schedule(job=configurable_job, cron_schedule="0 0 * * *")
def configurable_job_schedule(context: ScheduleEvaluationContext):
    scheduled_date = context.scheduled_execution_time.strftime("%Y-%m-%d")
    return RunRequest(
        run_key=None,
        run_config={
            "ops": {"configurable_op": {"config": {"scheduled_date": scheduled_date}}}
        },
        tags={"date": scheduled_date},
    )

To test this schedule, we used build_schedule_context to construct a ScheduleEvaluationContext to provide to the context parameter:

from dagster import build_schedule_context, validate_run_config


def test_configurable_job_schedule():
    context = build_schedule_context(
        scheduled_execution_time=datetime.datetime(2020, 1, 1)
    )
    run_request = configurable_job_schedule(context)
    assert validate_run_config(configurable_job, run_request.run_config)

If your @schedule-decorated function doesn't have a context parameter, you don't need to provide one when invoking it.

Testing schedules with resources#

For schedules that utilize resources, you can provide the resources when invoking the schedule function.

Let's say we want to test the process_data_schedule in this example:

from dagster import (
    schedule,
    ScheduleEvaluationContext,
    ConfigurableResource,
    job,
    RunRequest,
    RunConfig,
    Definitions,
)
from datetime import datetime
from typing import List

class DateFormatter(ConfigurableResource):
    format: str

    def strftime(self, dt: datetime) -> str:
        return dt.strftime(self.format)

@job
def process_data(): ...

@schedule(job=process_data, cron_schedule="* * * * *")
def process_data_schedule(
    context: ScheduleEvaluationContext,
    date_formatter: DateFormatter,
):
    formatted_date = date_formatter.strftime(context.scheduled_execution_time)

    return RunRequest(
        run_key=None,
        tags={"date": formatted_date},
    )

defs = Definitions(
    jobs=[process_data],
    schedules=[process_data_schedule],
    resources={"date_formatter": DateFormatter(format="%Y-%m-%d")},
)

In the test for this schedule, we provided the date_formatter resource to the schedule when we invoked its function:

from dagster import build_schedule_context, validate_run_config

def test_process_data_schedule():
    context = build_schedule_context(
        scheduled_execution_time=datetime.datetime(2020, 1, 1)
    )
    run_request = process_data_schedule(
        context, date_formatter=DateFormatter(format="%Y-%m-%d")
    )
    assert (
        run_request.run_config["ops"]["fetch_data"]["config"]["date"]
        == "2020-01-01"
    )

APIs in this guide#

Related#