docstr: Integration test of CoordinatorSenderNodeProvider.
docstr_tok: ["Integration","test","of","CoordinatorSenderNodeProvider","."]
code: def testCoordinatorSenderNodeProvider(self):
      """Integration test of CoordinatorSenderNodeProvider."""
      cluster_config = {
          "cluster_name": "random_name",
          "min_workers": 0,
          "max_workers": 0,
          "provider": {
              "type": "local",
              "coordinator_address": self.coordinator_address,
          },
          "head_node": {},
          "worker_nodes": {},
      }
      provider_config = cluster_config["provider"]
      node_provider_1 = _get_node_provider(
          provider_config,
          cluster_config["cluster_name"],
          use_cache=False,
      )
      assert isinstance(node_provider_1, CoordinatorSenderNodeProvider)
  
      assert not node_provider_1.non_terminated_nodes({})
      assert not node_provider_1.is_running(self.list_of_node_ips[0])
      assert node_provider_1.is_terminated(self.list_of_node_ips[0])
      assert not node_provider_1.node_tags(self.list_of_node_ips[0])
      head_node_tags = {
          TAG_RAY_NODE_KIND: NODE_KIND_HEAD,
      }
      assert not node_provider_1.non_terminated_nodes(head_node_tags)
      head_node_tags[TAG_RAY_NODE_NAME] = "ray-{}-head".format(
          cluster_config["cluster_name"]
      )
      node_provider_1.create_node(
          cluster_config["head_node"], head_node_tags, 1
      )
      assert node_provider_1.non_terminated_nodes({}) == [
          self.list_of_node_ips[0]
      ]
      head_node_tags[TAG_RAY_CLUSTER_NAME] = cluster_config[
          "cluster_name"
      ]
      assert (
          node_provider_1.node_tags(self.list_of_node_ips[0])
          == head_node_tags
      )
      assert node_provider_1.is_running(self.list_of_node_ips[0])
      assert not node_provider_1.is_terminated(self.list_of_node_ips[0])
  
      # Add another cluster.
      cluster_config["cluster_name"] = "random_name_2"
      provider_config = cluster_config["provider"]
      node_provider_2 = _get_node_provider(
          provider_config,
          cluster_config["cluster_name"],
          use_cache=False,
      )
      assert not node_provider_2.non_terminated_nodes({})
      assert not node_provider_2.is_running(self.list_of_node_ips[1])
      assert node_provider_2.is_terminated(self.list_of_node_ips[1])
      assert not node_provider_2.node_tags(self.list_of_node_ips[1])
      assert not node_provider_2.non_terminated_nodes(head_node_tags)
      head_node_tags[TAG_RAY_NODE_NAME] = "ray-{}-head".format(
          cluster_config["cluster_name"]
      )
      node_provider_2.create_node(
          cluster_config["head_node"], head_node_tags, 1
      )
      assert node_provider_2.non_terminated_nodes({}) == [
          self.list_of_node_ips[1]
      ]
      head_node_tags[TAG_RAY_CLUSTER_NAME] = cluster_config[
          "cluster_name"
      ]
      assert (
          node_provider_2.node_tags(self.list_of_node_ips[1])
          == head_node_tags
      )
      assert node_provider_2.is_running(self.list_of_node_ips[1])
      assert not node_provider_2.is_terminated(self.list_of_node_ips[1])
  
      # Add another cluster (should fail because we only have two nodes).
      cluster_config["cluster_name"] = "random_name_3"
      provider_config = cluster_config["provider"]
      node_provider_3 = _get_node_provider(
          provider_config,
          cluster_config["cluster_name"],
          use_cache=False,
      )
      assert not node_provider_3.non_terminated_nodes(head_node_tags)
      head_node_tags[TAG_RAY_NODE_NAME] = "ray-{}-head".format(
          cluster_config["cluster_name"]
      )
      node_provider_3.create_node(
          cluster_config["head_node"], head_node_tags, 1
      )
      assert not node_provider_3.non_terminated_nodes({})
  
      # Terminate all nodes.
      node_provider_1.terminate_node(self.list_of_node_ips[0])
      assert not node_provider_1.non_terminated_nodes({})
      node_provider_2.terminate_node(self.list_of_node_ips[1])
      assert not node_provider_2.non_terminated_nodes({})
  
      # Check if now we can create more clusters/nodes.
      node_provider_3.create_node(
          cluster_config["head_node"], head_node_tags, 1
      )
      worker_node_tags = {
          TAG_RAY_NODE_NAME: "ray-{}-worker".format(
              cluster_config["cluster_name"]
          ),
          TAG_RAY_NODE_KIND: NODE_KIND_WORKER,
      }
      node_provider_3.create_node(
          cluster_config["worker_nodes"], worker_node_tags, 1
      )
      assert (
          node_provider_3.non_terminated_nodes({})
          == self.list_of_node_ips
      )
      worker_filter = {TAG_RAY_NODE_KIND: NODE_KIND_WORKER}
      assert node_provider_3.non_terminated_nodes(worker_filter) == [
          self.list_of_node_ips[1]
      ]
      head_filter = {TAG_RAY_NODE_KIND: NODE_KIND_HEAD}
      assert node_provider_3.non_terminated_nodes(head_filter) == [
          self.list_of_node_ips[0]
      ]
  
imports: import os
  import random
  import unittest
  import socket
  import json
  from ray.autoscaler.local.coordinator_server import OnPremCoordinatorServer
  from ray.autoscaler._private.providers import _NODE_PROVIDERS, _get_node_provider
  from ray.autoscaler._private.local import config
  from ray.autoscaler._private.local.node_provider import LocalNodeProvider
  from ray.autoscaler._private.local.node_provider import record_local_head_state_if_needed
  from ray.autoscaler._private.local.coordinator_node_provider import CoordinatorSenderNodeProvider
  from ray.autoscaler.tags import TAG_RAY_NODE_KIND, TAG_RAY_CLUSTER_NAME, TAG_RAY_NODE_NAME, NODE_KIND_WORKER, NODE_KIND_HEAD, TAG_RAY_USER_NODE_TYPE, TAG_RAY_NODE_STATUS, STATUS_UP_TO_DATE
  from ray._private.utils import get_ray_temp_dir
  import pytest
label_desc: Use image node_id 8 for calling the OnPremCoordinatorServerTest obj's underlying member method code with example usage: obj.testCoordinatorSenderNodeProvider() without return types