code: def test_cf_predictions_extractor(
      self,
      eval_shared_model_names,
      model_specs,
      cf_configs,
      expected_predictions,
  ):
      model_path = self._makeIdentityParsingModel()
      eval_config = config_pb2.EvalConfig(model_specs=model_specs)
      eval_shared_models = []
      for model_name in eval_shared_model_names:
          eval_shared_models.append(
              model_eval_lib.default_eval_shared_model(
                  eval_saved_model_path=model_path,
                  tags=[tf.saved_model.SERVING],
                  model_name=model_name,
                  eval_config=eval_config,
              )
          )
      schema = text_format.Parse(
          """
          feature {
            name: "x"
            type: INT
          }
          feature {
            name: "x_cf1"
            type: INT
          }
          feature {
            name: "x_cf2"
            type: INT
          }
          """,
          schema_pb2.Schema(),
      )
  
      tfx_io = test_util.InMemoryTFExampleRecord(
          schema=schema,
          raw_record_column_name=constants.ARROW_INPUT_COLUMN,
      )
      examples = [
          self._makeExample(x=1, x_cf1=1, x_cf2=2),
          self._makeExample(x=2, x_cf1=1, x_cf2=2),
      ]
      num_examples = len(examples)
      tensor_adapter_config = tensor_adapter.TensorAdapterConfig(
          arrow_schema=tfx_io.ArrowSchema(),
          tensor_representations=tfx_io.TensorRepresentations(),
      )
      feature_extractor = features_extractor.FeaturesExtractor(
          eval_config=eval_config,
          tensor_representations=tensor_adapter_config.tensor_representations,
      )
  
      cf_predictions_extractor = counterfactual_predictions_extractor.CounterfactualPredictionsExtractor(
          eval_shared_models=eval_shared_models,
          eval_config=eval_config,
          cf_configs=cf_configs,
      )
  
      with beam.Pipeline() as pipeline:
          result = (
              pipeline
              | "Create"
              >> beam.Create(
                  [e.SerializeToString() for e in examples],
                  reshuffle=False,
              )
              | "BatchExamples"
              >> tfx_io.BeamSource(batch_size=num_examples)
              | "InputsToExtracts"
              >> model_eval_lib.BatchedInputsToExtracts()
              | "ExtractFeatures" >> feature_extractor.ptransform
              | cf_predictions_extractor.stage_name
              >> cf_predictions_extractor.ptransform
          )
  
          def check_result(got):
              try:
                  self.assertLen(got, 1)
                  # exact outputs are non-deterministic because model is trained
                  # so we can't assert full extracts. Instead, we just assert keys.
                  self.assertIn(constants.PREDICTIONS_KEY, got[0])
                  np.testing.assert_equal(
                      got[0][constants.PREDICTIONS_KEY],
                      expected_predictions,
                      err_msg=(
                          f"actual:{got[0][constants.PREDICTIONS_KEY]}\n"
                          f"expected:{expected_predictions}"
                      ),
                  )
                  self.assertNotIn(
                      counterfactual_predictions_extractor._TEMP_ORIG_INPUT_KEY,
                      got[0],
                  )
              except AssertionError as err:
                  raise util.BeamAssertException(err)
  
          util.assert_that(result, check_result, label="result")
  
imports: import os
  import tempfile
  from absl.testing import parameterized
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import tensorflow
  from tensorflow_model_analysis import constants
  from tensorflow_model_analysis.api import model_eval_lib
  from tensorflow_model_analysis.api import types
  from tensorflow_model_analysis.eval_saved_model import testutil
  from tensorflow_model_analysis.extractors import counterfactual_predictions_extractor
  from tensorflow_model_analysis.extractors import features_extractor
  from tensorflow_model_analysis.proto import config_pb2
  from tfx_bsl.tfxio import tensor_adapter
  from tfx_bsl.tfxio import test_util
  from google.protobuf import text_format
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 3 for calling the CounterfactualPredictionsExtactorTest obj's underlying member method code with example usage: obj.test_cf_predictions_extractor(eval_shared_model_names, model_specs, cf_configs, expected_predictions) without return types