dag_filepath = TEST_DAG_FOLDER / "test_on_failure_callback_dag.py"
        with create_session() as session:
            callback = DagCallbackRequest(
                dag_id="test_start_date_scheduling",
                full_filepath=str(dag_filepath),
                is_failure_callback=True,
                processor_subdir=str(tmpdir),
                run_id="123",
            )
            session.add(DbCallbackRequest(callback=callback, priority_weight=10))
        child_pipe, parent_pipe = multiprocessing.Pipe()
        manager = DagFileProcessorManager(
            dag_directory=tmpdir,
            max_runs=1,
            processor_timeout=timedelta(days=365),
            signal_conn=child_pipe,
            dag_ids=[],
            pickle_dags=False,
            async_mode=False,
        )
        with create_session() as session:
            results = self.run_processor_manager_one_loop(manager, parent_pipe)
        assert (len(results)) == 0
        
        with create_session() as session:
            assert session.query(DbCallbackRequest).count() == 1
    def test_callback_queue(self, tmpdir):