code: def _verify_transform_outputs(
      self,
      materialize=True,
      store_cache=True,
      multiple_example_inputs=False,
      disable_statistics=False,
      verify_sharded_statistics=False,
  ):
      expected_outputs = ["transformed_graph"]
  
      if store_cache:
          expected_outputs.append("CACHE")
          self.assertNotEqual(
              0,
              len(
                  fileio.listdir(
                      self._updated_analyzer_cache_artifact.uri
                  )
              ),
          )
  
      example_artifacts = self._example_artifacts[:1]
      transformed_example_artifacts = (
          self._transformed_example_artifacts[:1]
      )
      if multiple_example_inputs:
          example_artifacts = self._example_artifacts
          transformed_example_artifacts = (
              self._transformed_example_artifacts
          )
  
      if materialize:
          expected_outputs.append("transformed_examples")
  
          assert len(example_artifacts) == len(
              transformed_example_artifacts
          )
          for example, transformed_example in zip(
              example_artifacts, transformed_example_artifacts
          ):
              examples_train_files = fileio.glob(
                  os.path.join(example.uri, "Split-train", "*")
              )
              transformed_train_files = fileio.glob(
                  os.path.join(
                      transformed_example.uri, "Split-train", "*"
                  )
              )
              self.assertGreater(len(transformed_train_files), 0)
  
              examples_eval_files = fileio.glob(
                  os.path.join(example.uri, "Split-eval", "*")
              )
              transformed_eval_files = fileio.glob(
                  os.path.join(
                      transformed_example.uri, "Split-eval", "*"
                  )
              )
              self.assertGreater(len(transformed_eval_files), 0)
  
              # Construct datasets and count number of records in each split.
              examples_train_count = self._get_dataset_size(
                  examples_train_files
              )
              transformed_train_count = self._get_dataset_size(
                  transformed_train_files
              )
              examples_eval_count = self._get_dataset_size(
                  examples_eval_files
              )
              transformed_eval_count = self._get_dataset_size(
                  transformed_eval_files
              )
  
              # Check for each split that it contains the same number of records in
              # the input artifact as in the output artifact (i.e 1-to-1 mapping is
              # preserved).
              self.assertEqual(
                  examples_train_count, transformed_train_count
              )
              self.assertEqual(
                  examples_eval_count, transformed_eval_count
              )
              self.assertGreater(
                  transformed_train_count, transformed_eval_count
              )
  
      if disable_statistics:
          self.assertFalse(
              fileio.exists(
                  os.path.join(
                      self._pre_transform_schema.uri,
                      executor._SCHEMA_FILE,
                  )
              )
          )
          self.assertFalse(
              fileio.exists(
                  os.path.join(
                      self._post_transform_schema.uri,
                      executor._SCHEMA_FILE,
                  )
              )
          )
          self.assertFalse(
              fileio.exists(
                  os.path.join(
                      self._pre_transform_stats.uri, executor.STATS_FILE
                  )
              )
          )
          self.assertFalse(
              fileio.exists(
                  os.path.join(
                      self._post_transform_stats.uri,
                      executor.STATS_FILE,
                  )
              )
          )
          self.assertFalse(
              fileio.exists(
                  os.path.join(
                      self._post_transform_anomalies.uri,
                      executor._ANOMALIES_FILE,
                  )
              )
          )
  
      else:
          expected_outputs.extend(
              [
                  "pre_transform_schema",
                  "pre_transform_stats",
                  "post_transform_schema",
                  "post_transform_stats",
                  "post_transform_anomalies",
              ]
          )
          self.assertTrue(
              fileio.exists(
                  os.path.join(
                      self._pre_transform_schema.uri,
                      executor._SCHEMA_FILE,
                  )
              )
          )
          self.assertTrue(
              fileio.exists(
                  os.path.join(
                      self._post_transform_schema.uri,
                      executor._SCHEMA_FILE,
                  )
              )
          )
          self.assertTrue(
              fileio.exists(
                  os.path.join(
                      self._pre_transform_stats.uri, executor.STATS_FILE
                  )
              )
          )
          self.assertTrue(
              fileio.exists(
                  os.path.join(
                      self._post_transform_stats.uri,
                      executor.STATS_FILE,
                  )
              )
          )
          self.assertTrue(
              fileio.exists(
                  os.path.join(
                      self._post_transform_anomalies.uri,
                      executor._ANOMALIES_FILE,
                  )
              )
          )
      if verify_sharded_statistics:
          pre_transform_paths = fileio.glob(
              os.path.join(
                  self._pre_transform_stats.uri,
                  executor.SHARDED_STATS_PREFIX + "*",
              )
          )
          post_transform_paths = fileio.glob(
              os.path.join(
                  self._post_transform_stats.uri,
                  executor.SHARDED_STATS_PREFIX + "*",
              )
          )
  
          self.assertNotEmpty(pre_transform_paths)
          self.assertNotEmpty(post_transform_paths)
  
      # Depending on `materialize` and `store_cache`, check that
      # expected outputs are exactly correct. If either flag is False, its
      # respective output should not be present.
      self.assertCountEqual(
          expected_outputs, fileio.listdir(self._output_data_dir)
      )
  
      path_to_saved_model = os.path.join(
          self._transformed_output.uri,
          tft.TFTransformOutput.TRANSFORM_FN_DIR,
          tf.saved_model.SAVED_MODEL_FILENAME_PB,
      )
      self.assertTrue(fileio.exists(path_to_saved_model))
  
imports: import copy
  import json
  import os
  import re
  import tempfile
  from unittest import mock
  from absl.testing import parameterized
  import apache_beam
  import tensorflow
  import tensorflow_data_validation
  import tensorflow_transform
  from tensorflow_transform.beam import tft_unit
  from tfx import types
  from tfx.components.testdata.module_file import transform_module
  from tfx.components.transform import executor
  from tfx.dsl.io import fileio
  from tfx.proto import example_gen_pb2
  from tfx.proto import transform_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.types import standard_component_specs
  from tfx.utils import io_utils
  from tfx.utils import name_utils
  from tfx.utils import proto_utils
label_desc: Use image node_id 8 for calling the ExecutorTest obj's underlying member method code with example usage: obj._verify_transform_outputs(materialize, store_cache, multiple_example_inputs, disable_statistics, verify_sharded_statistics) without return types