dag_task2.set_downstream(dag_task3)
        initial_task_states = {
            "test_short_circuit_false": TaskInstanceState.SUCCESS,
            "test_state_skipped1": TaskInstanceState.SKIPPED,
            "test_state_skipped2": TaskInstanceState.SKIPPED,
        }
        dag_run = self.create_dag_run(dag=dag, task_states=initial_task_states, session=session)
        dag_run.update_state()
        assert DagRunState.SUCCESS == dag_run.state
    def test_dagrun_not_stuck_in_running_when_all_tasks_instances_are_removed(self, session):
        """
        Tests that a DAG run succeeds when all tasks are removed
        """
        dag = DAG(dag_id="test_dagrun_success_when_all_skipped", start_date=timezone.datetime(2017, 1, 1))
        dag_task1 = ShortCircuitOperator(
            task_id="test_short_circuit_false", dag=dag, python_callable=lambda: False
        )
        dag_task2 = EmptyOperator(task_id="test_state_skipped1", dag=dag)
        dag_task3 = EmptyOperator(task_id="test_state_skipped2", dag=dag)
        dag_task1.set_downstream(dag_task2)
        dag_task2.set_downstream(dag_task3)
        initial_task_states = {
            "test_short_circuit_false": TaskInstanceState.REMOVED,
            "test_state_skipped1": TaskInstanceState.REMOVED,
            "test_state_skipped2": TaskInstanceState.REMOVED,
        }
        dag_run = self.create_dag_run(dag=dag, task_states=initial_task_states, session=session)
        dag_run.update_state()
        assert DagRunState.SUCCESS == dag_run.state
    def test_dagrun_success_conditions(self, session):