This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
db.clear_db_dags()
    db.clear_db_runs()
    db.clear_db_serialized_dags()
    db.clear_dag_specific_permissions()
class TestDagBag:
    def setup_class(self):
        self.empty_dir = mkdtemp()
        db_clean_up()
    def teardown_class(self):
        shutil.rmtree(self.empty_dir)
        db_clean_up()
    def test_get_existing_dag(self):
        """
        Test that we're able to parse some example DAGs and retrieve them
        """
        dagbag = models.DagBag(dag_folder=self.empty_dir, include_examples=True)
        some_expected_dag_ids = ["example_bash_operator", "example_branch_operator"]
        for dag_id in some_expected_dag_ids:
            dag = dagbag.get_dag(dag_id)
            assert dag is not None
            assert dag_id == dag.dag_id
        assert dagbag.size() >= 7
    def test_get_non_existing_dag(self):
        """
        test that retrieving a non existing dag id returns None without crashing
        """
        dagbag = models.DagBag(dag_folder=self.empty_dir, include_examples=False)
        non_existing_dag_id = "non_existing_dag_id"
        assert dagbag.get_dag(non_existing_dag_id) is None
    def test_serialized_dag_not_existing_doesnt_raise(self):