)
    def test_wait_for_downstream(self, session, prev_ti_state, is_ti_success):
        dag_id = "test_wait_for_downstream"
        dag = self.dagbag.get_dag(dag_id)
        upstream, downstream = dag.tasks
        
        
        dag_run_1 = self.create_dag_run(
            dag,
            execution_date=timezone.datetime(2016, 1, 1, 0, 0, 0),
            is_backfill=True,
            session=session,
        )
        dag_run_2 = self.create_dag_run(
            dag,
            execution_date=timezone.datetime(2016, 1, 2, 0, 0, 0),
            is_backfill=True,
            session=session,
        )
        prev_ti_downstream = TI(task=downstream, run_id=dag_run_1.run_id)
        ti = TI(task=upstream, run_id=dag_run_2.run_id)
        prev_ti = ti.get_previous_ti()
        prev_ti.set_state(TaskInstanceState.SUCCESS)
        assert prev_ti.state == TaskInstanceState.SUCCESS
        prev_ti_downstream.set_state(prev_ti_state)
        ti.set_state(TaskInstanceState.QUEUED)
        ti.run()
        assert (ti.state == TaskInstanceState.SUCCESS) == is_ti_success
    @pytest.mark.parametrize("state", [DagRunState.QUEUED, DagRunState.RUNNING])
    def test_next_dagruns_to_examine_only_unpaused(self, session, state):