### 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
"""
        Returns a map of task instance key to task instance object for the tasks to
        run in the given dag run.
        :param dag_run: the dag run to get the tasks from
        :type dag_run: airflow.models.DagRun
        :param session: the database session object
        :type session: sqlalchemy.orm.session.Session
        """
        tasks_to_run = {}
        if dag_run is None:
            return tasks_to_run
        
        self.reset_state_for_orphaned_tasks(filter_by_dag_run=dag_run, session=session)
        
        dag_run.refresh_from_db()
        make_transient(dag_run)
        try:
            for ti in dag_run.get_task_instances():
                
                if ti.state == State.NONE:
                    ti.set_state(State.SCHEDULED, session=session)
                if ti.state != State.REMOVED:
                    tasks_to_run[ti.key] = ti
            session.commit()
        except Exception:
            session.rollback()
            raise
        return tasks_to_run
    def _log_progress(self, ti_status):