test_start_date = timezone.utcnow() - datetime.timedelta(days=1)
        for i, callback in enumerate([[sla_callback], sla_callback]):
            dag, task = create_dummy_dag(
                dag_id=f"test_sla_miss_{i}",
                task_id="dummy",
                sla_miss_callback=callback,
                default_args={"start_date": test_start_date, "sla": datetime.timedelta(hours=1)},
            )
            mock_stats_incr.reset_mock()
            session.merge(TaskInstance(task=task, execution_date=test_start_date, state="Success"))
            
            session.merge(
                SlaMiss(task_id="dummy", dag_id=f"test_sla_miss_{i}", execution_date=test_start_date)
            )
            
            mock_log = mock.MagicMock()
            dag_file_processor = DagFileProcessor(dag_ids=[], dag_directory=TEST_DAGS_FOLDER, log=mock_log)
            dag_file_processor.manage_slas(dag=dag, session=session)
            assert sla_callback.called
            mock_log.exception.assert_called_once_with(
                "Could not call sla_miss_callback(%s) for DAG %s",
                sla_callback.__name__,
                f"test_sla_miss_{i}",
            )
            mock_stats_incr.assert_called_once_with(
                "sla_callback_notification_failure",
                tags={"dag_id": f"test_sla_miss_{i}", "func_name": sla_callback.__name__},
            )
    @mock.patch("airflow.dag_processing.processor.send_email")
    def test_dag_file_processor_only_collect_emails_from_sla_missed_tasks(