code: def test_cache_helpers_round_trip(self):
      base_test_dir = os.path.join(
          os.environ.get(
              "TEST_UNDECLARED_OUTPUTS_DIR", self.get_temp_dir()
          ),
          self._testMethodName,
      )
  
      dataset_key_0_metadata = analyzer_cache.DatasetCacheMetadata(42)
      dataset_key_1_metadata = analyzer_cache.DatasetCacheMetadata(17)
      dataset_key_0 = analyzer_cache.DatasetKey("dataset_key_0")
      dataset_key_1 = analyzer_cache.DatasetKey("dataset_key_1")
      dataset_keys = (dataset_key_0, dataset_key_1)
  
      with beam.Pipeline() as p:
          cache_pcoll_dict = {
              dataset_key_0: analyzer_cache.DatasetCache(
                  {
                      b"\x8a": p
                      | "CreateA" >> beam.Create([b"[1, 2, 3]"]),
                      b"\x8b": p | "CreateB" >> beam.Create([b"[5]"]),
                      b"\x8b1": p | "CreateB1" >> beam.Create([b"[6]"]),
                  },
                  p
                  | "CreateM0" >> beam.Create([dataset_key_0_metadata]),
              ),
              dataset_key_1: analyzer_cache.DatasetCache(
                  {
                      b"\x8c": p
                      | "CreateC" >> beam.Create([b"[9, 5, 2, 1]"]),
                  },
                  p
                  | "CreateM1" >> beam.Create([dataset_key_1_metadata]),
              ),
          }
  
          _ = cache_pcoll_dict | analyzer_cache.WriteAnalysisCacheToFS(
              p, base_test_dir, dataset_keys
          )
  
      with beam.Pipeline() as p:
          read_cache = p | analyzer_cache.ReadAnalysisCacheFromFS(
              base_test_dir,
              list(cache_pcoll_dict.keys()),
              [b"\x8a", b"\x8b", b"\x8c"],
          )
  
          beam_test_util.assert_that(
              read_cache[dataset_key_0].cache_dict[b"\x8a"],
              beam_test_util.equal_to([b"[1, 2, 3]"]),
              label="AssertA",
          )
          beam_test_util.assert_that(
              read_cache[dataset_key_0].cache_dict[b"\x8b"],
              beam_test_util.equal_to([b"[5]"]),
              label="AssertB",
          )
          beam_test_util.assert_that(
              read_cache[dataset_key_0].metadata,
              beam_test_util.equal_to([dataset_key_0_metadata]),
              label="Assert0Size",
          )
          beam_test_util.assert_that(
              read_cache[dataset_key_1].cache_dict[b"\x8c"],
              beam_test_util.equal_to([b"[9, 5, 2, 1]"]),
              label="AssertC",
          )
          beam_test_util.assert_that(
              read_cache[dataset_key_1].metadata,
              beam_test_util.equal_to([dataset_key_1_metadata]),
              label="Assert1Size",
          )
  
file_path: transform/tensorflow_transform/beam/analyzer_cache_test.py
imports: import os
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import tensorflow
  from tensorflow_transform import analyzer_nodes
  from tensorflow_transform import analyzers
  from tensorflow_transform.beam import analyzer_cache
  from tensorflow_transform import test_case
label_desc: Use image node_id 3 for calling the AnalyzerCacheTest obj's underlying member method code with example usage: obj.test_cache_helpers_round_trip() without return types