docstr: If OnPremCoordinatorState __init__ generates correct state file.
  
  Test the general use case and if the coordinator server crashes or
  updates the list of node ips with more/less nodes.
docstr_tok: ["If","OnPremCoordinatorState","__init__","generates","correct","state","file",".","Test","the","general","use","case","and","if","the","coordinator","server","crashes","or","updates","the","list","of","node","ips","with","more\/less","nodes","."]
code: def testOnPremCoordinatorStateInit(self):
      """If OnPremCoordinatorState __init__ generates correct state file.
  
      Test the general use case and if the coordinator server crashes or
      updates the list of node ips with more/less nodes.
      """
  
      expected_nodes = {}
      for ip in self.list_of_node_ips:
          expected_nodes[ip] = {
              "tags": {},
              "state": "terminated",
          }
  
      state_save_path = "/tmp/coordinator.state"
      assert os.path.exists(state_save_path)
      nodes = json.loads(open(state_save_path).read())
      assert nodes == expected_nodes
  
      # Test removing workers updates the cluster state.
      del expected_nodes[self.list_of_node_ips[1]]
      self.server.shutdown()
      self.server = OnPremCoordinatorServer(
          list_of_node_ips=self.list_of_node_ips[0:1],
          host=self.host,
          port=self.port,
      )
      nodes = json.loads(open(state_save_path).read())
      assert nodes == expected_nodes
  
      # Test adding back workers updates the cluster state.
      expected_nodes[self.list_of_node_ips[1]] = {
          "tags": {},
          "state": "terminated",
      }
      self.server.shutdown()
      self.server = OnPremCoordinatorServer(
          list_of_node_ips=self.list_of_node_ips,
          host=self.host,
          port=self.port,
      )
      nodes = json.loads(open(state_save_path).read())
      assert nodes == expected_nodes
  
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 7 for calling the OnPremCoordinatorServerTest obj's underlying member method code with example usage: obj.testOnPremCoordinatorStateInit() without return types