dagbag = DagBag(dag_folder="/dev/null", include_examples=True, read_dags_from_db=False)
        dag_file_processor = DagFileProcessor(
            dag_ids=[], dag_directory=TEST_DAGS_FOLDER, log=mock.MagicMock()
        )
        dag_file_processor.UNIT_TEST_MODE = False
        with create_session() as session:
            dag = dagbag.get_dag("example_branch_operator")
            task = dag.get_task(task_id="run_this_first")
            dagrun = dag.create_dagrun(
                state=State.RUNNING,
                execution_date=DEFAULT_DATE,
                run_type=DagRunType.SCHEDULED,
                session=session,
            )
            ti = TaskInstance(task, run_id=dagrun.run_id, state=State.RUNNING)
            ti.hostname = "test_hostname"
            session.add(ti)
        requests = [
            TaskCallbackRequest(
                full_filepath="A", simple_task_instance=SimpleTaskInstance.from_ti(ti), msg="Message"
            )
        ]
        dag_file_processor.execute_callbacks(dagbag, requests)
        with create_session() as session:
            tis = session.query(TaskInstance)
            assert tis[0].hostname == "test_hostname"
    def test_process_file_should_failure_callback(self, monkeypatch, tmp_path, get_test_dag):