'Test that DaskExecutor can be used to backfill example dags'

@unittest.skipIf(SKIP_DASK, 'Dask    unsupported    by    this    configuration')
def test_backfill_integration(self):

    cluster = LocalCluster()
    dags = [dag for dag in self.dagbag.dags.values() if (dag.dag_id in ['example_bash_operator'])]
    for dag in dags:
        dag.clear(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
    for (i, dag) in enumerate(sorted(dags, key=(lambda d: d.dag_id))):
        job = BackfillJob(dag=dag, start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_first_depends_on_past=True, executor=DaskExecutor(cluster_address=cluster.scheduler_address))
        job.run()
    cluster.close()
