session.merge(
            SlaMiss(task_id=task.task_id, dag_id=dag.dag_id, execution_date=now - datetime.timedelta(days=2))
        )
        session.flush()
        dag_file_processor = DagFileProcessor(
            dag_ids=[], dag_directory=TEST_DAGS_FOLDER, log=mock.MagicMock()
        )
        dag_file_processor.manage_slas(dag=dag, session=session)
        sla_miss_count = (
            session.query(SlaMiss)
            .filter(
                SlaMiss.dag_id == dag.dag_id,
                SlaMiss.task_id == task.task_id,
            )
            .count()
        )
        assert sla_miss_count == 2
        mock_stats_incr.assert_called_with(
            "sla_missed", tags={"dag_id": "test_sla_miss", "run_id": "test", "task_id": "dummy"}
        )
    @mock.patch("airflow.dag_processing.processor.Stats.incr")
    def test_dag_file_processor_sla_miss_callback_exception(self, mock_stats_incr, create_dummy_dag):
        """
        Test that the dag file processor gracefully logs an exception if there is a problem
        calling the sla_miss_callback
        """
        session = settings.Session()
        sla_callback = MagicMock(
            __name__="function_name", side_effect=RuntimeError("Could not call function")
        )