try:
            fail_stop_dag.add_task(task_with_default_trigger_rule)
        except DagInvalidTriggerRule as exception:
            assert (
                False
            ), f"dag.add_task() raises exception for fail-stop dag & default trigger rule: {exception}"
        
        with pytest.raises(DagInvalidTriggerRule):
            fail_stop_dag.add_task(task_with_non_default_trigger_rule)
    def test_dag_add_task_sets_default_task_group(self):
        dag = DAG(dag_id="test_dag_add_task_sets_default_task_group", start_date=DEFAULT_DATE)
        task_without_task_group = EmptyOperator(task_id="task_without_group_id")
        default_task_group = TaskGroupContext.get_current_task_group(dag)
        dag.add_task(task_without_task_group)
        assert default_task_group.get_child_by_label("task_without_group_id") == task_without_task_group
        task_group = TaskGroup(group_id="task_group", dag=dag)
        task_with_task_group = EmptyOperator(task_id="task_with_task_group", task_group=task_group)
        dag.add_task(task_with_task_group)
        assert task_group.get_child_by_label("task_with_task_group") == task_with_task_group
        assert dag.get_task("task_group.task_with_task_group") == task_with_task_group
    @pytest.mark.parametrize("dag_run_state", [DagRunState.QUEUED, DagRunState.RUNNING])
    def test_clear_set_dagrun_state(self, dag_run_state):
        dag_id = "test_clear_set_dagrun_state"
        self._clean_up(dag_id)
        task_id = "t1"
        dag = DAG(dag_id, start_date=DEFAULT_DATE, max_active_runs=1)
        t_1 = EmptyOperator(task_id=task_id, dag=dag)
        session = settings.Session()
        dagrun_1 = dag.create_dagrun(
            run_type=DagRunType.BACKFILL_JOB,