filename_to_parse = tmpdir / "temp_dag.py"
        dag_code = dedent(
            """
        from airflow import DAG
        dag = DAG(dag_id='temp_dag', schedule='0 0 * * *')
        """
        )
        with open(filename_to_parse, "w") as file_to_parse:
            file_to_parse.writelines(dag_code)
        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,
        )
        self.run_processor_manager_one_loop(manager, parent_pipe)
        last_runtime = manager.get_last_runtime(manager.file_paths[0])
        child_pipe.close()
        parent_pipe.close()
        statsd_timing_mock.assert_called_with(
            "dag_processing.last_duration.temp_dag", timedelta(seconds=last_runtime)
        )
    def test_refresh_dags_dir_doesnt_delete_zipped_dags(self, tmpdir):