docstr: Tests skipped nodes have status msg updates when generating tasks.
docstr_tok: ["Tests","skipped","nodes","have","status","msg","updates","when","generating","tasks","."]
code: def test_task_generation_when_node_skipped(self):
      """Tests skipped nodes have status msg updates when generating tasks."""
  
      with mock.patch.object(
          task_gen_utils, "generate_resolved_info", autospec=True
      ) as mock_generate_resolved_info:
          mock_generate_resolved_info.side_effect = (
              exceptions.InputResolutionError()
          )
          expected_error = (
              "failure to resolve inputs; node uid:"
              " NodeUid(pipeline_uid=PipelineUid(pipeline_id='my_pipeline',"
              " pipeline_run_id=None), node_id='my_transform'); error:"
              " ['tfx.orchestration.portable.input_resolution.exceptions.InputResolutionError\\n']"
          )
          tasks = self._generate_and_test(
              use_task_queue=False,
              num_initial_executions=0,
              num_tasks_generated=3,
              num_new_executions=0,
              num_active_executions=0,
          )
  
          self.assertIsInstance(tasks[0], task_lib.UpdateNodeStateTask)
          self.assertEqual(pstate.NodeState.STARTED, tasks[1].state)
          self.assertEqual(
              status_lib.Code.UNAVAILABLE, tasks[2].status.code
          )
          self.assertEqual(expected_error, tasks[1].status.message)
  
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 8 for calling the AsyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_task_generation_when_node_skipped() without return types