serialized_dag_count = (
                session.query(func.count(SerializedDagModel.dag_id))
                .filter(SerializedDagModel.fileloc == test_dag_path)
                .scalar()
            )
            assert serialized_dag_count == 0
    @conf_vars(
        {
            ("core", "load_examples"): "False",
            ("scheduler", "standalone_dag_processor"): "True",
        }
    )
    def test_deactivate_stale_dags_standalone_mode(self):
        """
        Ensure only dags from current dag_directory are updated
        """
        dag_directory = "directory"
        manager = DagFileProcessorManager(
            dag_directory=dag_directory,
            max_runs=1,
            processor_timeout=timedelta(minutes=10),
            signal_conn=MagicMock(),
            dag_ids=[],
            pickle_dags=False,
            async_mode=True,
        )
        test_dag_path = str(TEST_DAG_FOLDER / "test_example_bash_operator.py")
        dagbag = DagBag(test_dag_path, read_dags_from_db=False)
        other_test_dag_path = str(TEST_DAG_FOLDER / "test_scheduler_dags.py")
        other_dagbag = DagBag(other_test_dag_path, read_dags_from_db=False)