docstr: Tests stopped nodes are ignored when generating tasks.
docstr_tok: ["Tests","stopped","nodes","are","ignored","when","generating","tasks","."]
code: def test_task_generation_when_node_stopped(self, stop_transform):
      """Tests stopped nodes are ignored when generating tasks."""
      # Simulate that ExampleGen has already completed successfully.
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 1, 1
      )
  
      # Generate once.
      num_initial_executions = 1
      if stop_transform:
          num_tasks_generated = 2
          num_new_executions = 0
          num_active_executions = 0
          with self._mlmd_connection as m:
              pipeline_state = test_utils.get_or_create_pipeline_state(
                  m, self._pipeline
              )
              with pipeline_state:
                  with pipeline_state.node_state_update_context(
                      task_lib.NodeUid.from_node(
                          self._pipeline, self._transform
                      )
                  ) as node_state:
                      node_state.update(
                          pstate.NodeState.STOPPING,
                          status_lib.Status(
                              code=status_lib.Code.CANCELLED
                          ),
                      )
      else:
          num_tasks_generated = 4
          num_new_executions = 1
          num_active_executions = 1
      tasks = self._generate_and_test(
          True,
          num_initial_executions=num_initial_executions,
          num_tasks_generated=num_tasks_generated,
          num_new_executions=num_new_executions,
          num_active_executions=num_active_executions,
      )
      self.assertLen(tasks, num_tasks_generated)
  
      if stop_transform:
          self.assertIsInstance(tasks[0], task_lib.UpdateNodeStateTask)
          self.assertEqual(pstate.NodeState.RUNNING, tasks[0].state)
      else:
          self.assertIsInstance(tasks[0], task_lib.UpdateNodeStateTask)
          self.assertEqual(pstate.NodeState.RUNNING, tasks[0].state)
          self.assertIsInstance(tasks[1], task_lib.UpdateNodeStateTask)
          self.assertEqual(pstate.NodeState.RUNNING, tasks[1].state)
          self.assertIsInstance(tasks[2], task_lib.ExecNodeTask)
  
imports: import os
  from absl.testing import parameterized
  from absl.testing.absltest import mock
  import tensorflow
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import async_pipeline_task_gen
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import service_jobs
  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_async_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.utils import status
label_desc: Use image node_id 7 for calling the AsyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_task_generation_when_node_stopped(stop_transform) without return types