code: def test_backfill_pure_service_node(self):
      backfill_token = "backfill-20230227-180505-123456"
      test_utils.get_or_create_pipeline_state(
          self._mlmd_connection, self._pipeline
      )
      # Put ExampleGen in backfill mode.
      with pstate.PipelineState.load(
          self._mlmd_connection,
          task_lib.PipelineUid.from_pipeline(self._pipeline),
      ) as pipeline_state:
          example_gen_node = task_lib.NodeUid.from_node(
              self._pipeline,
              node_proto_view.get_view(self._example_gen),
          )
          with pipeline_state.node_state_update_context(
              example_gen_node
          ) as node_state:
              node_state.update(
                  pstate.NodeState.STARTING,
                  backfill_token=backfill_token,
              )
      # Generate a RUNNING task for ExampleGen backfill.
      [
          running_example_gen_task,
          update_transform_task,
          update_trainer_task,
      ] = self._generate_and_test(
          use_task_queue=False,
          num_initial_executions=0,
          num_tasks_generated=3,
          num_new_executions=0,
          num_active_executions=0,
          expected_exec_nodes=[],
      )
  
      self.assertIsInstance(
          running_example_gen_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          running_example_gen_task.state, pstate.NodeState.RUNNING
      )
      self.assertIsInstance(
          update_transform_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          pstate.NodeState.STARTED, update_transform_task.state
      )
      self.assertIsInstance(
          update_trainer_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          pstate.NodeState.STARTED, update_trainer_task.state
      )
      self.assertEqual(
          running_example_gen_task.backfill_token, backfill_token
      )
      self._mock_service_job_manager.ensure_node_services.assert_has_calls(
          [
              mock.call(
                  mock.ANY,
                  self._example_gen.node_info.id,
                  backfill_token,
              ),
          ]
      )
  
      # Mark ExampleGen backfill service job as COMPLETED.
      def _backfill_completes(
          unused_pipeline_state, node_id, unused_backfill_token=""
      ):
          if node_id == self._example_gen.node_info.id:
              return service_jobs.ServiceStatus(
                  code=service_jobs.ServiceStatusCode.SUCCESS
              )
  
      self._mock_service_job_manager.reset_mock()
      self._mock_service_job_manager.ensure_node_services.side_effect = (
          _backfill_completes
      )
  
      # Generate a STOPPED task after ExampleGen backfill completes.
      [
          stopped_example_gen_task,
          update_transform_task,
          update_trainer_task,
      ] = self._generate_and_test(
          use_task_queue=False,
          num_initial_executions=0,
          num_tasks_generated=3,
          num_new_executions=0,
          num_active_executions=0,
          expected_exec_nodes=[],
      )
      self.assertIsInstance(
          stopped_example_gen_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          stopped_example_gen_task.state, pstate.NodeState.STOPPED
      )
      self.assertIsInstance(
          update_transform_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          pstate.NodeState.STARTED, update_transform_task.state
      )
      self.assertIsInstance(
          update_trainer_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          pstate.NodeState.STARTED, update_trainer_task.state
      )
      self.assertEqual(stopped_example_gen_task.backfill_token, "")
      self._mock_service_job_manager.ensure_node_services.assert_has_calls(
          [
              mock.call(
                  mock.ANY,
                  self._example_gen.node_info.id,
                  backfill_token,
              ),
          ]
      )
      self._mock_service_job_manager.stop_node_services.assert_called_once_with(
          mock.ANY, self._example_gen.node_info.id
      )
  
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 12 for calling the AsyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_backfill_pure_service_node() and returns: service_jobs