)
        with create_session() as session:
            session.add(DbCallbackRequest(callback=callback1, priority_weight=11))
            session.add(DbCallbackRequest(callback=callback2, 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:
            self.run_processor_manager_one_loop(manager, parent_pipe)
            assert session.query(DbCallbackRequest).count() == 1
    @conf_vars(
        {
            ("scheduler", "standalone_dag_processor"): "True",
            ("scheduler", "max_callbacks_per_loop"): "2",
            ("core", "load_examples"): "False",
        }
    )
    def test_fetch_callbacks_from_database_max_per_loop(self, tmpdir):
        """Test DagFileProcessorManager._fetch_callbacks method"""
        dag_filepath = TEST_DAG_FOLDER / "test_on_failure_callback_dag.py"
        with create_session() as session:
            for i in range(5):
                callback = DagCallbackRequest(