Writing TestsΒΆ

The workflows pytest fixture is registered automatically, but it depends upon the pytest-django test database fixture.

See usage examples in Executing DAGs.

To override the test database setup, for example to replace Django with another testing framework, you can implement two fixtures:

  • db should create a test database and tear it down afterwards.

  • conn should return a DBApi connection to that database with an open transaction that will be rolled back afterwards.

The test fixture provides these methods to make testing easier:

setup_task_inputs

This method loops through the tasks listed as inputs to the provided task. The arguments are (task, initial_data=None). For each input task it loops through the outputs and calls the test_setup method. If initial_data is provided, it should be a dictionary with keys matching output names and values representing data to pass to the corresponding test_setup method.

This provides a mechanism for creating upstream tables and inserting test data.

execute_task

This method is used to execute the task being tested. The arguments are (task, parameters=None). It creates a context dictionary to pass to task.execute(context), including a test database connection. After execution, it collects the data generated by looping through the outputs and calling test_results on each, collecting the data into a dictionary by output name.

execute_dag

This method executes a DAG or subset thereof, using the provided environment variables. The arguments are (dag, env_vars=None, begin=None, end=None).

execute_sql

This method runs arbitrary SQL against the test database. It can be used to manually setup input data or to assert outputs. There is a signle argument (sql) and if any rows are returned they are provided as a list of dicts.