code: def testResolveInputArtifacts(self):
      pipeline = self.load_pipeline_proto(
          "pipeline_for_input_resolver_test.pbtxt"
      )
      my_example_gen = pipeline.nodes[0].pipeline_node
      another_example_gen = pipeline.nodes[1].pipeline_node
      my_transform = pipeline.nodes[2].pipeline_node
      my_trainer = pipeline.nodes[3].pipeline_node
  
      with self.get_metadata() as m:
          # Publishes first ExampleGen with two output channels. `output_examples`
          # will be consumed by downstream Transform.
          output_example = self.make_examples(uri="my_examples_uri")
          side_examples = self.make_examples(uri="side_examples_uri")
          output_artifacts = self.fake_execute(
              m,
              my_example_gen,
              input_map=None,
              output_map={
                  "output_examples": [output_example],
                  "another_examples": [side_examples],
              },
          )
          output_example = output_artifacts["output_examples"][0]
          side_examples = output_artifacts["another_examples"][0]
          del side_examples
  
          # Publishes second ExampleGen with one output channel with the same output
          # key as the first ExampleGen. However this is not consumed by downstream
          # nodes.
          another_output_example = self.make_examples(
              uri="another_examples_uri"
          )
          self.fake_execute(
              m,
              another_example_gen,
              input_map=None,
              output_map={"output_examples": [another_output_example]},
          )
  
          # Gets inputs for transform. Should get back what the first ExampleGen
          # published in the `output_examples` channel.
          transform_inputs = inputs_utils.resolve_input_artifacts(
              metadata_handle=m, pipeline_node=my_transform
          )[0]
          self.assertArtifactMapEqual(
              {
                  "examples_1": [output_example],
                  "examples_2": [output_example],
              },
              transform_inputs,
          )
  
          # Tries to resolve inputs for trainer. As trainer also requires min_count
          # for both input channels (from example_gen and from transform) but we did
          # not publish anything from transform, it should return nothing.
          with self.assertRaisesRegex(
              exceptions.InsufficientInputError,
              "InputSpec min_count has not met",
          ):
              inputs_utils.resolve_input_artifacts(
                  metadata_handle=m, pipeline_node=my_trainer
              )
  
          # Tries to resolve inputs for transform after adding a new context query
          # to the input spec that refers to a non-existent context. Inputs cannot
          # be resolved in this case.
          context_query = (
              my_transform.inputs.inputs["examples_1"]
              .channels[0]
              .context_queries.add()
          )
          context_query.type.name = "non_existent_context"
          context_query.name.field_value.string_value = (
              "non_existent_context"
          )
          with self.assertRaisesRegex(
              exceptions.InsufficientInputError,
              "InputSpec min_count has not met",
          ):
              inputs_utils.resolve_input_artifacts(
                  metadata_handle=m, pipeline_node=my_transform
              )
  
imports: import collections
  import os
  import tensorflow
  from tfx import types
  from tfx.dsl.compiler import placeholder_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.portable import data_types
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable import inputs_utils
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import test_case_utils
  from google.protobuf import text_format
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 3 for calling the InputsUtilsTest obj's underlying member method code with example usage: obj.testResolveInputArtifacts() without return types