code: def test_install_ray_on_new_nodes(self):
      self.base_provider.create_node(
          {}, {TAG_RAY_NODE_KIND: "worker_nodes1"}, 1
      )
      instance = Instance(
          instance_id="0",
          instance_type="worker_nodes1",
          cloud_instance_id="0",
          status=Instance.ALLOCATED,
      )
      success, verison = self.instance_storage.upsert_instance(instance)
      assert success
      instance.version = verison
      self.runner.respond_to_call(
          "json .Config.Env", ["[]" for i in range(1)]
      )
  
      self.threaded_ray_installer._install_ray_on_new_nodes(
          instance.instance_id
      )
      self.threaded_ray_installer._ray_installation_executor.shutdown(
          wait=True
      )
      instances, _ = self.instance_storage.get_instances(
          instance_ids={instance.instance_id}
      )
      assert (
          instances[instance.instance_id].status == Instance.RAY_RUNNING
      )
  
file_path: ray/python/ray/autoscaler/v2/tests/test_threaded_ray_installer.py
imports: import os
  import sys
  import unittest
  from unittest.mock import patch
  import pytest
  from ray._private.test_utils import load_test_config
  from ray.autoscaler.tags import TAG_RAY_NODE_KIND
  from ray.autoscaler.v2.instance_manager.config import AutoscalingConfig
  from ray.autoscaler.v2.instance_manager.instance_storage import InstanceStorage
  from ray.autoscaler.v2.instance_manager.ray_installer import RayInstaller
  from ray.autoscaler.v2.instance_manager.storage import InMemoryStorage
  from ray.autoscaler.v2.instance_manager.subscribers.threaded_ray_installer import ThreadedRayInstaller
  from ray.core.generated.instance_manager_pb2 import Instance
  from ray.tests.autoscaler_test_utils import MockProcessRunner, MockProvider
label_desc: Use image node_id 4 for calling the ThreadedRayInstallerTest obj's underlying member method code with example usage: obj.test_install_ray_on_new_nodes() without return types