state=State.FAILED,
            start_date=DEFAULT_DATE,
            execution_date=DEFAULT_DATE,
        )
        session.merge(dagrun_1)
        task_instance_1 = TI(t_1, execution_date=DEFAULT_DATE, state=State.RUNNING)
        session.merge(task_instance_1)
        session.commit()
        dag.clear(
            start_date=DEFAULT_DATE,
            end_date=DEFAULT_DATE + datetime.timedelta(days=1),
            dag_run_state=dag_run_state,
            include_subdags=False,
            include_parentdag=False,
            session=session,
        )
        dagruns = (
            session.query(
                DagRun,
            )
            .filter(
                DagRun.dag_id == dag_id,
            )
            .all()
        )
        assert len(dagruns) == 1
        dagrun: DagRun = dagruns[0]
        assert dagrun.state == dag_run_state
    @pytest.mark.parametrize("dag_run_state", [DagRunState.QUEUED, DagRunState.RUNNING])
    def test_clear_set_dagrun_state_for_mapped_task(self, dag_run_state):