code: def testBuildDummyConsumerWithCondition(self):
      producer_task_1 = test_utils.dummy_producer_component(
          output1=channel_utils.as_channel(
              [standard_artifacts.Model()]
          ),
          param1="value1",
      ).with_id("producer_task_1")
      producer_task_2 = test_utils.dummy_producer_component_2(
          output1=channel_utils.as_channel(
              [standard_artifacts.Model()]
          ),
          param1="value2",
      ).with_id("producer_task_2")
      # This test tests two things:
      # 1. Nested conditions. The condition string of consumer_task should contain
      #    both predicates.
      # 2. Implicit channels. consumer_task only takes producer_task_1's output.
      #    But producer_task_2 is used in condition, hence producer_task_2 should
      #    be added to the dependency of consumer_task.
      # See testdata for detail.
      with conditional.Cond(
          producer_task_1.outputs["output1"].future()[0].uri != "uri"
      ):
          with conditional.Cond(
              producer_task_2.outputs["output1"]
              .future()[0]
              .property("property")
              == "value1"
          ):
              consumer_task = test_utils.dummy_consumer_component(
                  input1=producer_task_1.outputs["output1"],
                  param1=1,
              )
      # Need to construct a pipeline to set producer_component_id.
      pipeline = tfx.dsl.Pipeline(
          pipeline_name="pipeline-with-condition",
          pipeline_root="",
          components=[producer_task_1, producer_task_2, consumer_task],
      )
      pipeline.finalize()
      deployment_config = pipeline_pb2.PipelineDeploymentConfig()
      component_defs = {}
      my_builder = step_builder.StepBuilder(
          node=consumer_task,
          image="gcr.io/tensorflow/tfx:latest",
          deployment_config=deployment_config,
          component_defs=component_defs,
          dsl_context_reg=pipeline.dsl_context_registry,
      )
      actual_step_spec = self._sole(my_builder.build())
      actual_component_def = self._sole(component_defs)
  
      self.assertProtoEquals(
          test_utils.get_proto_from_test_data(
              "expected_dummy_consumer_with_condition_component.pbtxt",
              pipeline_pb2.ComponentSpec(),
          ),
          actual_component_def,
      )
      self.assertProtoEquals(
          test_utils.get_proto_from_test_data(
              "expected_dummy_consumer_with_condition_task.pbtxt",
              pipeline_pb2.PipelineTaskSpec(),
          ),
          actual_step_spec,
      )
      self.assertProtoEquals(
          test_utils.get_proto_from_test_data(
              "expected_dummy_consumer_with_condition_executor.pbtxt",
              pipeline_pb2.PipelineDeploymentConfig(),
          ),
          deployment_config,
      )
  
imports: from typing import Any, Dict
  from kfp.pipeline_spec import pipeline_spec_pb2
  import tensorflow
  from tfx import components
  from tfx import v1
  from tfx.dsl.components.common import importer
  from tfx.dsl.components.common import resolver
  from tfx.dsl.context_managers import dsl_context_registry
  from tfx.dsl.experimental.conditionals import conditional
  from tfx.dsl.input_resolution.strategies import latest_artifact_strategy
  from tfx.dsl.input_resolution.strategies import latest_blessed_model_strategy
  from tfx.extensions.google_cloud_big_query.example_gen import component
  from tfx.orchestration import data_types
  from tfx.orchestration.kubeflow import decorators
  from tfx.orchestration.kubeflow.v2 import parameter_utils
  from tfx.orchestration.kubeflow.v2 import step_builder
  from tfx.orchestration.kubeflow.v2 import test_utils
  from tfx.proto import example_gen_pb2
  from tfx.types import channel
  from tfx.types import channel_utils
  from tfx.types import standard_artifacts
label_desc: Use image node_id 14 for calling the StepBuilderTest obj's underlying member method code with example usage: obj.testBuildDummyConsumerWithCondition() without return types