### Body

Add unit test for changes in https://github.com/apache/airflow/pull/17305

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
airflow/jobs/backfill_job.py
dag_run = self._get_dag_run(dagrun_info, dag, session=session)
                tis_map = self._task_instances_for_dag_run(dag_run, session=session)
                if dag_run is None:
                    continue
                ti_status.active_runs.append(dag_run)
                ti_status.to_run.update(tis_map or {})
        processed_dag_run_dates = self._process_backfill_task_instances(
            ti_status=ti_status,
            executor=executor,
            pickle_id=pickle_id,
            start_date=start_date,
            session=session,
        )
        ti_status.executed_dag_run_dates.update(processed_dag_run_dates)
    @provide_session
    def _set_unfinished_dag_runs_to_failed(self, dag_runs, session=None):
        """
        Go through the dag_runs and update the state based on the task_instance state.
        Then set DAG runs that are not finished to failed.
        :param dag_runs: DAG runs
        :param session: session
        :return: None
        """
        for dag_run in dag_runs:
            dag_run.update_state()
            if dag_run.state not in State.finished:
                dag_run.set_state(State.FAILED)
            session.merge(dag_run)
    @provide_session
    def _execute(self, session=None):