code: def testModelSignaturesDoFn(
      self,
      save_as_keras,
      extract_key_and_signature_names,
      default_signature_names,
      prefer_dict_outputs,
      use_schema,
      expected_num_outputs,
  ):
      export_path = self.createModelWithMultipleDenseInputs(
          save_as_keras
      )
      eval_shared_models = {}
      model_specs = []
      for sigs in extract_key_and_signature_names.values():
          for model_name in sigs:
              if model_name not in eval_shared_models:
                  eval_shared_models[
                      model_name
                  ] = self.createTestEvalSharedModel(
                      eval_saved_model_path=export_path,
                      model_name=model_name,
                      tags=[tf.saved_model.SERVING],
                  )
                  model_specs.append(
                      config_pb2.ModelSpec(name=model_name)
                  )
      schema = self.createDenseInputsSchema() if use_schema else None
      tfx_io = tf_example_record.TFExampleBeamRecord(
          physical_format="text",
          schema=schema,
          raw_record_column_name=constants.ARROW_INPUT_COLUMN,
      )
  
      examples = [
          self._makeExample(input_1=1.0, input_2=2.0),
          self._makeExample(input_1=3.0, input_2=4.0),
          self._makeExample(input_1=5.0, input_2=6.0),
      ]
      assert len(extract_key_and_signature_names) == 1
      extract_key, signature_names = next(
          iter(extract_key_and_signature_names.items())
      )
      with beam.Pipeline() as pipeline:
          # pylint: disable=no-value-for-parameter
          result = (
              pipeline
              | "Create"
              >> beam.Create([e.SerializeToString() for e in examples])
              | "BatchExamples" >> tfx_io.BeamSource(batch_size=3)
              | "ToExtracts" >> beam.Map(_record_batch_to_extracts)
              | "ModelSignatures"
              >> beam.ParDo(
                  model_util.ModelSignaturesDoFn(
                      model_specs=model_specs,
                      eval_shared_models=eval_shared_models,
                      output_keypath=[extract_key],
                      signature_names=signature_names,
                      default_signature_names=default_signature_names,
                      prefer_dict_outputs=prefer_dict_outputs,
                  )
              )
          )
  
          # pylint: enable=no-value-for-parameter
  
          def check_result(got):
              try:
                  self.assertLen(got, 1)
                  for key in extract_key_and_signature_names:
                      self.assertIn(key, got[0])
                      if prefer_dict_outputs:
                          self.assertIsInstance(got[0][key], dict)
                          self.assertEqual(
                              tfma_util.batch_size(got[0][key]),
                              expected_num_outputs,
                          )
  
              except AssertionError as err:
                  raise util.BeamAssertException(err)
  
          util.assert_that(result, check_result, label="result")
  
imports: import tempfile
  import unittest
  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 types
  from tensorflow_model_analysis.eval_saved_model import testutil
  from tensorflow_model_analysis.proto import config_pb2
  from tensorflow_model_analysis.utils import model_util
  from tensorflow_model_analysis.utils import util
  from tfx_bsl.tfxio import tf_example_record
  from google.protobuf import text_format
  from tensorflow.core.protobuf import saved_model_pb2
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 24 for calling the ModelUtilTest obj's underlying member method code with example usage: obj.testModelSignaturesDoFn(save_as_keras, extract_key_and_signature_names, default_signature_names, prefer_dict_outputs, use_schema, expected_num_outputs) without return types