code: def test_resolver_task_scheduler(self):
      with self._mlmd_connection as m:
          # Publishes two models which will be consumed by downstream resolver.
          output_model_1 = types.Artifact(
              self._trainer.outputs.outputs["model"].artifact_spec.type
          )
          output_model_1.uri = "my_model_uri_1"
  
          output_model_2 = types.Artifact(
              self._trainer.outputs.outputs["model"].artifact_spec.type
          )
          output_model_2.uri = "my_model_uri_2"
  
          contexts = context_lib.prepare_contexts(
              m, self._trainer.contexts
          )
          execution = execution_publish_utils.register_execution(
              m, self._trainer.node_info.type, contexts
          )
          execution_publish_utils.publish_succeeded_execution(
              m,
              execution.id,
              contexts,
              {
                  "model": [output_model_1, output_model_2],
              },
          )
  
      task_queue = tq.TaskQueue()
  
      # Verify that resolver task is generated.
      [resolver_task] = test_utils.run_generator_and_test(
          test_case=self,
          mlmd_connection_manager=self._mlmd_cm,
          generator_class=sptg.SyncPipelineTaskGenerator,
          pipeline=self._pipeline,
          task_queue=task_queue,
          use_task_queue=False,
          service_job_manager=None,
          num_initial_executions=1,
          num_tasks_generated=1,
          num_new_executions=1,
          num_active_executions=1,
          expected_exec_nodes=[self._resolver_node],
          ignore_update_node_state_tasks=True,
      )
  
      with self._mlmd_connection as m:
          # Run resolver task scheduler and publish results.
          ts_result = resolver_task_scheduler.ResolverTaskScheduler(
              mlmd_handle=m, pipeline=self._pipeline, task=resolver_task
          ).schedule()
          self.assertEqual(status_lib.Code.OK, ts_result.status.code)
          self.assertIsInstance(
              ts_result.output, task_scheduler.ResolverNodeOutput
          )
          self.assertCountEqual(
              ["resolved_model"],
              ts_result.output.resolved_input_artifacts.keys(),
          )
          models = ts_result.output.resolved_input_artifacts[
              "resolved_model"
          ]
          self.assertLen(models, 1)
          self.assertEqual(
              "my_model_uri_2", models[0].mlmd_artifact.uri
          )
          post_execution_utils.publish_execution_results_for_task(
              m, resolver_task, ts_result
          )
  
      # Verify resolver node output is input to the downstream consumer node.
      [consumer_task] = test_utils.run_generator_and_test(
          test_case=self,
          mlmd_connection_manager=self._mlmd_cm,
          generator_class=sptg.SyncPipelineTaskGenerator,
          pipeline=self._pipeline,
          task_queue=task_queue,
          use_task_queue=False,
          service_job_manager=None,
          num_initial_executions=2,
          num_tasks_generated=1,
          num_new_executions=1,
          num_active_executions=1,
          expected_exec_nodes=[self._consumer_node],
          ignore_update_node_state_tasks=True,
      )
      self.assertCountEqual(
          ["resolved_model"], consumer_task.input_artifacts.keys()
      )
      input_models = consumer_task.input_artifacts["resolved_model"]
      self.assertLen(input_models, 1)
      self.assertEqual(
          "my_model_uri_2", input_models[0].mlmd_artifact.uri
      )
  
imports: import os
  import uuid
  import tensorflow
  from tfx import types
  from tfx.dsl.compiler import constants
  from tfx.orchestration.experimental.core import post_execution_utils
  from tfx.orchestration.experimental.core import sync_pipeline_task_gen
  from tfx.orchestration.experimental.core import task_queue
  from tfx.orchestration.experimental.core import task_scheduler
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.task_schedulers import resolver_task_scheduler
  from tfx.orchestration.experimental.core.testing import test_pipeline_with_resolver
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable import runtime_parameter_utils
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.utils import status
label_desc: Use image node_id 3 for calling the ResolverTaskSchedulerTest obj's underlying member method code with example usage: obj.test_resolver_task_scheduler() without return types