docstr: Tests conditionals in the pipeline.
  
  Args:
    evaluate: Whether to run the conditional evaluator.
docstr_tok: ["Tests","conditionals","in","the","pipeline",".","Args",":","evaluate",":","Whether","to","run","the","conditional","evaluator","."]
code: def test_conditional_execution(self, evaluate):
      """Tests conditionals in the pipeline.
  
      Args:
        evaluate: Whether to run the conditional evaluator.
      """
      # Check the expected terminal nodes.
      layers = sptg._topsorted_layers(self._pipeline)
      self.assertEqual(
          {
              self._example_validator.node_info.id,
              self._chore_b.node_info.id,
              self._evaluator.node_info.id,
          },
          sptg._terminal_node_ids(layers, {}),
      )
  
      # Start executing the pipeline:
  
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 1, 1
      )
  
      self._run_next(False, expect_nodes=[self._stats_gen])
      self._run_next(False, expect_nodes=[self._schema_gen])
      self._run_next(
          False, expect_nodes=[self._example_validator, self._transform]
      )
  
      # Evaluator is run conditionally based on whether the Model artifact
      # produced by the trainer has a custom property evaluate=1.
      self._run_next(
          False,
          expect_nodes=[self._trainer],
          artifact_custom_properties={"evaluate": 1}
          if evaluate
          else None,
      )
  
      tasks = self._generate(False)
      [evaluator_update_node_state_task] = [
          t
          for t in tasks
          if isinstance(t, task_lib.UpdateNodeStateTask)
          and t.node_uid.node_id == "my_evaluator"
      ]
      self.assertEqual(
          pstate.NodeState.RUNNING
          if evaluate
          else pstate.NodeState.SKIPPED,
          evaluator_update_node_state_task.state,
      )
  
      exec_node_tasks = [
          t for t in tasks if isinstance(t, task_lib.ExecNodeTask)
      ]
      if evaluate:
          [
              chore_a_exec_node_task,
              evaluator_exec_node_task,
          ] = exec_node_tasks
          self.assertEqual(
              "chore_a", chore_a_exec_node_task.node_uid.node_id
          )
          self.assertEqual(
              "my_evaluator", evaluator_exec_node_task.node_uid.node_id
          )
          self._finish_node_execution(False, chore_a_exec_node_task)
          self._finish_node_execution(False, evaluator_exec_node_task)
      else:
          [chore_a_exec_node_task] = exec_node_tasks
          self.assertEqual(
              "chore_a", chore_a_exec_node_task.node_uid.node_id
          )
          self._finish_node_execution(False, chore_a_exec_node_task)
  
      self._run_next(False, expect_nodes=[self._chore_b])
  
      # All nodes executed, finalization task should be produced.
      [finalize_task] = self._generate(False, True)
      self.assertIsInstance(
          finalize_task, task_lib.FinalizePipelineTask
      )
  
imports: import itertools
  import os
  from typing import Literal
  import uuid
  from absl.testing import parameterized
  from absl.testing.absltest import mock
  import tensorflow
  from tfx.dsl.compiler import constants
  from tfx.orchestration import data_types_utils
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import mlmd_state
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import service_jobs
  from tfx.orchestration.experimental.core import sync_pipeline_task_gen
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import task_queue
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_sync_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import runtime_parameter_utils
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import status
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 21 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_conditional_execution(evaluate) without return types