This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
"""
        Test that Serialized DAG is updated in DagBag when it is updated in
        Serialized DAG table after 'min_serialized_dag_fetch_interval' seconds are passed.
        """
        with time_machine.travel((tz.datetime(2020, 1, 5, 0, 0, 0)), tick=False):
            example_bash_op_dag = DagBag(include_examples=True).dags.get("example_bash_operator")
            SerializedDagModel.write_dag(dag=example_bash_op_dag)
            dag_bag = DagBag(read_dags_from_db=True)
            ser_dag_1 = dag_bag.get_dag("example_bash_operator")
            ser_dag_1_update_time = dag_bag.dags_last_fetched["example_bash_operator"]
            assert example_bash_op_dag.tags == ser_dag_1.tags
            assert ser_dag_1_update_time == tz.datetime(2020, 1, 5, 0, 0, 0)
        
        
        with time_machine.travel((tz.datetime(2020, 1, 5, 0, 0, 4)), tick=False):
            with assert_queries_count(0):
                assert dag_bag.get_dag("example_bash_operator").tags == ["example", "example2"]
        
        with time_machine.travel((tz.datetime(2020, 1, 5, 0, 0, 6)), tick=False):
            example_bash_op_dag.tags += ["new_tag"]
            SerializedDagModel.write_dag(dag=example_bash_op_dag)
        
        
        with time_machine.travel((tz.datetime(2020, 1, 5, 0, 0, 8)), tick=False):
            with assert_queries_count(2):
                updated_ser_dag_1 = dag_bag.get_dag("example_bash_operator")
                updated_ser_dag_1_update_time = dag_bag.dags_last_fetched["example_bash_operator"]
        assert set(updated_ser_dag_1.tags) == {"example", "example2", "new_tag"}
        assert updated_ser_dag_1_update_time > ser_dag_1_update_time
    def test_collect_dags_from_db(self):