code: def _verify_exec_node_task(
      test_case,
      pipeline,
      node,
      execution_id,
      task,
      expected_context_names,
  ):
      """Verifies that generated ExecNodeTask has the expected properties for the node."""
      if not expected_context_names:
          expected_context_names = [
              "my_pipeline",
              f"my_pipeline.{node.node_info.id}",
          ]
      test_case.assertEqual(
          task_lib.NodeUid.from_node(pipeline, node), task.node_uid
      )
      test_case.assertEqual(execution_id, task.execution_id)
      if pipeline.execution_mode == pipeline_pb2.Pipeline.SYNC:
          expected_context_names.append(
              pipeline.runtime_spec.pipeline_run_id.field_value.string_value
          )
      expected_input_artifacts_keys = [
          key
          for key, value in node.inputs.inputs.items()
          if not value.hidden
      ]
      expected_output_artifacts_keys = list(
          iter(node.outputs.outputs.keys())
      )
      if expected_output_artifacts_keys:
          output_artifact_uri = os.path.join(
              pipeline.runtime_spec.pipeline_root.field_value.string_value,
              node.node_info.id,
              expected_output_artifacts_keys[0],
              str(execution_id),
          )
          test_case.assertEqual(
              output_artifact_uri,
              task.output_artifacts[expected_output_artifacts_keys[0]][
                  0
              ].uri,
          )
      # There may be cached context which we ignore.
      test_case.assertContainsSubset(
          expected_context_names, [c.name for c in task.contexts]
      )
      test_case.assertCountEqual(
          expected_input_artifacts_keys,
          list(task.input_artifacts.keys()),
      )
      test_case.assertCountEqual(
          expected_output_artifacts_keys,
          list(task.output_artifacts.keys()),
      )
      test_case.assertEqual(
          os.path.join(
              pipeline.runtime_spec.pipeline_root.field_value.string_value,
              node.node_info.id,
              ".system",
              "executor_execution",
              str(execution_id),
              "executor_output.pb",
          ),
          task.executor_output_uri,
      )
      test_case.assertEqual(
          os.path.join(
              pipeline.runtime_spec.pipeline_root.field_value.string_value,
              node.node_info.id,
              ".system",
              "stateful_working_dir",
              _MOCKED_STATEFUL_WORKING_DIR_INDEX,
          ),
          task.stateful_working_dir,
      )
  
imports: import os
  from typing import Dict, Optional
  import uuid
  from absl.testing.absltest import mock
  from tfx import types
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import env
  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 task
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import cache_utils
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import standard_artifacts
  from tfx.utils import status
  from tfx.utils import test_case_utils
  from tfx.utils import typing_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 19 for calling a global function with example usage: _verify_exec_node_task(test_case, pipeline, node, execution_id, task, expected_context_names) without return types