docstr: Tests behavior of a dynamic active_filter.
docstr_tok: ["Tests","behavior","of","a","dynamic","active_filter","."]
code: def testActiveFilter_dynamicFilterBehavior(self):
      """Tests behavior of a dynamic active_filter."""
      loader_registry = []
      loader_factory = functools.partial(
          _TimestampedByteLoader, registry=loader_registry
      )
      threshold = 0
      active_filter = lambda timestamp: timestamp >= threshold
      self._loader = directory_loader.DirectoryLoader(
          self._directory, loader_factory, active_filter=active_filter
      )
  
      def assertLoadersForPaths(paths):
          paths = [
              os.path.join(self._directory, path) for path in paths
          ]
          self.assertEqual(loader_registry, paths)
  
      self._WriteToFile("a", ["A1", "A2"], [1, 2])
      self._WriteToFile("b", ["B1", "B2", "B3"], [1, 2, 3])
      self._WriteToFile("c", ["C1"], [1])
      threshold = 2
      # First load pass should leave file C marked inactive.
      self.assertLoaderYields(["A1", "A2", "B1", "B2", "B3", "C1"])
      assertLoadersForPaths(["a", "b"])
      self._WriteToFile("a", ["A4"], [4])
      self._WriteToFile("b", ["B4"], [4])
      self._WriteToFile("c", ["C4"], [4])
      threshold = 3
      # Second load pass should mark file A as inactive (due to newly
      # increased threshold) and thus skip reading data from it.
      self.assertLoaderYields(["B4"])
      assertLoadersForPaths(["b"])
      self._WriteToFile("b", ["B5", "B6"], [5, 6])
      # Simulate a third pass in which the threshold increases while
      # we're processing a file, so it's still active at the start of the
      # load but should be marked inactive at the end.
      load_generator = self._loader.Load()
      self.assertEqual("B5", next(load_generator))
      threshold = 7
      self.assertEqual(["B6"], list(load_generator))
      assertLoadersForPaths([])
      # Confirm that all loaders are now inactive.
      self._WriteToFile("b", ["B7"], [7])
      self.assertLoaderYields([])
  
imports: import functools
  import glob
  import os
  import shutil
  from unittest import mock
  import tensorflow
  from tensorboard.backend.event_processing import directory_loader
  from tensorboard.backend.event_processing import directory_watcher
  from tensorboard.backend.event_processing import event_file_loader
  from tensorboard.backend.event_processing import io_wrapper
  from tensorboard.util import test_util
label_desc: Use image node_id 11 for calling the DirectoryLoaderTest obj's underlying member method code with example usage: obj.testActiveFilter_dynamicFilterBehavior() without return types