code: def testConditionals(self):
      a1, a2, a3, a4 = self.create_artifacts(4)
  
      # Set blessed custom property
      a1.set_int_custom_property("blessed", 1)
      a2.set_int_custom_property("blessed", 0)
      a3.set_int_custom_property("blessed", 0)
      a4.set_int_custom_property("blessed", 1)
  
      # Set tag custom property
      a1.set_string_custom_property("tag", "foo")
      a2.set_string_custom_property("tag", "foo")
      a3.set_string_custom_property("tag", "bar")
      a4.set_string_custom_property("tag", "bar")
  
      x = self.parse_input_spec(
          """
        input_graph_ref {
          graph_id: "graph_1"
          key: "a"
        }
      """
      )
      graph_1 = self.parse_input_graph(
          """
        nodes {
          key: "op_1"
          value {
            op_node {
              op_type: "DummyOp"
            }
            output_data_type: ARTIFACT_MULTIMAP_LIST
          }
        }
        result_node: "op_1"
      """
      )
      self.mock_graph_fn_result(
          graph_1,
          lambda _: [
              {"a": [a1]},
              {"a": [a2]},
              {"a": [a3]},
              {"a": [a4]},
          ],
      )
  
      # Only allows artifact.custom_properties['blessed'] == 1,
      # which is a1 and a4.
      is_blessed = channel_utils.encode_placeholder_with_channels(
          DummyChannel("x").future()[0].custom_property("blessed") == 1,
          lambda channel: channel.name,
      )
  
      # Only allows artifact.custom_properties['tag'] == 'foo'
      # which is a1 and a2.
      is_foo = channel_utils.encode_placeholder_with_channels(
          (
              DummyChannel("x").future()[0].custom_property("tag")
              == "foo"
          ),
          lambda channel: channel.name,
      )
  
      cond_1 = pipeline_pb2.NodeInputs.Conditional(
          placeholder_expression=is_blessed
      )
      cond_2 = pipeline_pb2.NodeInputs.Conditional(
          placeholder_expression=is_foo
      )
  
      with self.subTest("blessed == 1"):
          node_inputs = pipeline_pb2.NodeInputs(
              inputs={"x": x},
              input_graphs={"graph_1": graph_1},
              conditionals={"cond_1": cond_1},
          )
  
          result = node_inputs_resolver.resolve(
              self._mlmd_handle, node_inputs
          )
          self.assertEqual(result, [{"x": [a1]}, {"x": [a4]}])
  
      with self.subTest("blessed == 1 and tag == foo"):
          node_inputs = pipeline_pb2.NodeInputs(
              inputs={"x": x},
              input_graphs={"graph_1": graph_1},
              conditionals={"cond_1": cond_1, "cond_2": cond_2},
          )
  
          result = node_inputs_resolver.resolve(
              self._mlmd_handle, node_inputs
          )
          self.assertEqual(result, [{"x": [a1]}])
  
imports: from typing import Set
  from unittest import mock
  import tensorflow
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.orchestration.portable.input_resolution import input_graph_resolver
  from tfx.orchestration.portable.input_resolution import node_inputs_resolver
  from tfx.orchestration.portable.input_resolution import partition_utils
  from tfx.orchestration.portable.input_resolution import channel_resolver
  from tfx.proto.orchestration import pipeline_pb2
  import tfx.types
  from tfx.types import channel_utils
  from tfx.utils import test_case_utils
  from google.protobuf import text_format
label_desc: Use image node_id 17 for calling the NodeInputsResolverTest obj's underlying member method code with example usage: obj.testConditionals() without return types