callback_file = tmp_path.joinpath("callback.txt")
        callback_file.touch()
        monkeypatch.setenv("AIRFLOW_CALLBACK_FILE", str(callback_file))
        dag_file_processor = DagFileProcessor(
            dag_ids=[], dag_directory=TEST_DAGS_FOLDER, log=mock.MagicMock()
        )
        dag = get_test_dag("test_on_failure_callback")
        task = dag.get_task(task_id="test_on_failure_callback_task")
        with create_session() as session:
            dagrun = dag.create_dagrun(
                state=State.RUNNING,
                execution_date=DEFAULT_DATE,
                run_type=DagRunType.SCHEDULED,
                session=session,
            )
            ti = dagrun.get_task_instance(task.task_id)
            ti.refresh_from_task(task)
            requests = [
                TaskCallbackRequest(
                    full_filepath=dag.fileloc,
                    simple_task_instance=SimpleTaskInstance.from_ti(ti),
                    msg="Message",
                )
            ]
            dag_file_processor.process_file(dag.fileloc, requests, session=session)
        ti.refresh_from_db()
        msg = " ".join([str(k) for k in ti.key.primary]) + " fired callback"
        assert msg in callback_file.read_text()
    @conf_vars({("core", "dagbag_import_error_tracebacks"): "False"})
    def test_add_unparseable_file_before_sched_start_creates_import_error(self, tmpdir):