self, mock_send_email, create_dummy_dag
    ):
        session = settings.Session()
        test_start_date = timezone.utcnow() - datetime.timedelta(days=1)
        email1 = "test1@test.com"
        dag, task = create_dummy_dag(
            dag_id="test_sla_miss",
            task_id="sla_missed",
            email=email1,
            default_args={"start_date": test_start_date, "sla": datetime.timedelta(hours=1)},
        )
        session.merge(TaskInstance(task=task, execution_date=test_start_date, state="Success"))
        email2 = "test2@test.com"
        EmptyOperator(task_id="sla_not_missed", dag=dag, owner="airflow", email=email2)
        session.merge(SlaMiss(task_id="sla_missed", dag_id="test_sla_miss", execution_date=test_start_date))
        dag_file_processor = DagFileProcessor(
            dag_ids=[], dag_directory=TEST_DAGS_FOLDER, log=mock.MagicMock()
        )
        dag_file_processor.manage_slas(dag=dag, session=session)
        assert len(mock_send_email.call_args_list) == 1
        send_email_to = mock_send_email.call_args_list[0][0][0]
        assert email1 in send_email_to
        assert email2 not in send_email_to
    @mock.patch("airflow.dag_processing.processor.Stats.incr")
    @mock.patch("airflow.utils.email.send_email")
    def test_dag_file_processor_sla_miss_email_exception(