docstr: Tests that nodes with only skipped downstream nodes are terminal nodes.
docstr_tok: ["Tests","that","nodes","with","only","skipped","downstream","nodes","are","terminal","nodes","."]
code: def test_terminal_nodes_with_partial_run(self):
      """Tests that nodes with only skipped downstream nodes are terminal nodes."""
      # Check the expected skipped and terminal nodes.
      self._example_gen.execution_options.skip.SetInParent()
      self._chore_a.execution_options.skip.SetInParent()
      self._chore_b.execution_options.skip.SetInParent()
      self._evaluator.execution_options.skip.SetInParent()
  
      with self._mlmd_connection as m:
          pipeline_state = test_utils.get_or_create_pipeline_state(
              m, self._pipeline
          )
          with pipeline_state:
              node_states_dict = pipeline_state.get_node_states_dict()
      expected_skipped_node_ids = {
          "my_example_gen",
          "chore_a",
          "chore_b",
          "my_evaluator",
      }
      self.assertEqual(
          expected_skipped_node_ids,
          sptg._skipped_node_ids(node_states_dict),
      )
  
      layers = sptg._topsorted_layers(self._pipeline)
      # All downstream nodes of trainer are marked as skipped, so it's considered
      # a terminal node.
      self.assertEqual(
          {
              self._trainer.node_info.id,
              self._example_validator.node_info.id,
          },
          sptg._terminal_node_ids(layers, expected_skipped_node_ids),
      )
  
      # Start executing the pipeline:
      test_utils.fake_cached_example_gen_run(
          self._mlmd_connection, self._example_gen
      )
      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]
      )
      self._run_next(False, expect_nodes=[self._trainer])
      # All runnable 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 11 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_terminal_nodes_with_partial_run() without return types