'Helper for testing DagRun states with simple two-task DAGS.
This is hackish: a dag run is created but its tasks are
run by a backfill.'

@provide_session
def evaluate_dagrun(self, dag_id, expected_task_states, dagrun_state, run_kwargs=None, advance_execution_date=False, session=None):

    if (run_kwargs is None):
        run_kwargs = {}
    scheduler = SchedulerJob(**self.default_scheduler_args)
    dag = self.dagbag.get_dag(dag_id)
    dag.clear()
    dr = scheduler.create_dag_run(dag)
    if advance_execution_date:
        dr = scheduler.create_dag_run(dag)
    ex_date = dr.execution_date
    try:
        dag.run(start_date=ex_date, end_date=ex_date, **run_kwargs)
    except AirflowException:
        pass
    for (task_id, expected_state) in expected_task_states.items():
        task = dag.get_task(task_id)
        ti = TI(task, ex_date)
        ti.refresh_from_db()
        self.assertEqual(ti.state, expected_state)
    dr = DagRun.find(dag_id=dag_id, execution_date=ex_date)
    dr = dr[0]
    dr.dag = dag
    self.assertEqual(dr.state, dagrun_state)
