filename_to_parse = tmpdir / "temp_dag.py"
        
        with open(filename_to_parse, "w") as file_to_parse:
            file_to_parse.writelines("an invalid airflow DAG")
        child_pipe, parent_pipe = multiprocessing.Pipe()
        async_mode = "sqlite" not in conf.get("database", "sql_alchemy_conn")
        manager = DagFileProcessorManager(
            dag_directory=tmpdir,
            max_runs=1,
            processor_timeout=timedelta(days=365),
            signal_conn=child_pipe,
            dag_ids=[],
            pickle_dags=False,
            async_mode=async_mode,
        )
        with create_session() as session:
            self.run_processor_manager_one_loop(manager, parent_pipe)
            import_errors = session.query(errors.ImportError).all()
            assert len(import_errors) == 1
            filename_to_parse.remove()
            
            self.run_processor_manager_one_loop(manager, parent_pipe)
            import_errors = session.query(errors.ImportError).all()
            assert len(import_errors) == 0
            session.rollback()
        child_pipe.close()
        parent_pipe.close()
    @conf_vars({("core", "load_examples"): "False"})
    def test_max_runs_when_no_files(self):