nonlocal fake_processors
            processor = FakeDagFileProcessorRunner._create_process(*args, **kwargs)
            fake_processors.append(processor)
            return processor
        mock_processor.side_effect = fake_processor_
        manager = DagFileProcessorManager(
            dag_directory=dag_filepath,
            dag_ids=[],
            
            max_runs=100,
            processor_timeout=timedelta(seconds=5),
            signal_conn=child_pipe,
            pickle_dags=False,
            async_mode=True,
        )
        try:
            thread.start()
            
            manager._run_parsing_loop()
            exit_event.set()
        finally:
            logging.info("Closing pipes")
            parent_pipe.close()
            child_pipe.close()
            thread.join(timeout=1.0)
    @conf_vars({("core", "load_examples"): "False"})
    @mock.patch("airflow.dag_processing.manager.Stats.timing")
    def test_send_file_processing_statsd_timing(self, statsd_timing_mock, tmpdir):