class_docstr: Utility to test BatchingNodeProvider.
docstr: Calls non_terminated_nodes and validates output against this test classes's
  accumulated expected state.
  
  Tests methods internal_ip, node_tags, non_terminated_nodes of
  BatchingNodeProvider.
docstr_tok: ["Calls","non_terminated_nodes","and","validates","output","against","this","test","classes's","accumulated","expected","state",".","Tests","methods","internal_ip",",","node_tags",",","non_terminated_nodes","of","BatchingNodeProvider","."]
code: def validate_non_terminated_nodes(self):
      """Calls non_terminated_nodes and validates output against this test classes's
      accumulated expected state.
  
      Tests methods internal_ip, node_tags, non_terminated_nodes of
      BatchingNodeProvider.
      """
      nodes = self.node_provider.non_terminated_nodes({})
  
      # Get counts of nodes using provider.node_tags to extract node type.
      actual_node_counts = defaultdict(int)
      for node in nodes:
          # Trivial check. Just confirming we can call internal_ip with no issue.
          assert isinstance(self.node_provider.internal_ip(node), str)
  
          # Check tag structure.
          tags = self.node_provider.node_tags(node)
          assert set(tags.keys()) == {
              TAG_RAY_USER_NODE_TYPE,
              TAG_RAY_NODE_STATUS,
              TAG_RAY_NODE_KIND,
          }
          node_type = tags[TAG_RAY_USER_NODE_TYPE]
          node_kind = tags[TAG_RAY_NODE_KIND]
          node_status = tags[TAG_RAY_NODE_STATUS]
          if node_type == "head":
              assert node_kind == NODE_KIND_HEAD
          else:
              assert node_kind == NODE_KIND_WORKER
          # Just by construction of this test:
          assert node_status == STATUS_UP_TO_DATE
  
          actual_node_counts[node_type] += 1
  
      # Remove 0 values from expected_node_counts before comparing.
      for k, v in copy(self.expected_node_counts).items():
          if v == 0:
              del self.expected_node_counts[k]
      assert actual_node_counts == self.expected_node_counts
  
      # Get node counts again using tag filters.
      actual_node_counts_again = {}
      for node_type in actual_node_counts:
          actual_node_counts_again[node_type] = len(
              self.node_provider.non_terminated_nodes(
                  tag_filters={TAG_RAY_USER_NODE_TYPE: node_type}
              )
          )
      assert actual_node_counts_again == self.expected_node_counts
  
      # Check filtering by node kind.
      workers = self.node_provider.non_terminated_nodes(
          tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}
      )
      heads = self.node_provider.non_terminated_nodes(
          tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_HEAD}
      )
      assert len(heads) == 1
      assert set(nodes) == set(workers) | set(heads)
  
      # Check filtering by status.
      up_to_date_nodes = self.node_provider.non_terminated_nodes(
          tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE}
      )
      assert set(up_to_date_nodes) == set(nodes)
  
      # Make some assertions about internal structure of the node provider.
      expected_node_counts_without_head = copy(
          self.expected_node_counts
      )
      del expected_node_counts_without_head["head"]
      # Desired number of workers immediately after calling non_terminated_nodes is
      # current number of workers.
      assert (
          self.node_provider.scale_request.desired_num_workers
          == expected_node_counts_without_head
      )
      # scale_change_needed should be reset after calling non_terminated_nodes
      # (meaning: we've just obtained cluster state and have no indication
      # from create_node or terminate_node calls that scale needs to change.)
      assert self.node_provider.scale_change_needed is False
      # We've submitted the expected number of scale requests:
      assert (
          self.node_provider._scale_request_submitted_count
          == self.expected_scale_request_submitted_count
      )
  
imports: from copy import copy
  from uuid import uuid4
  import os
  import random
  import sys
  from typing import Any, Dict
  from collections import defaultdict
  import pytest
  from ray.autoscaler.batching_node_provider import BatchingNodeProvider, NodeData, ScaleRequest
  from ray.autoscaler._private.util import NodeID, NodeType
  from ray.autoscaler.tags import STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE, TAG_RAY_NODE_KIND, TAG_RAY_NODE_STATUS, NODE_KIND_HEAD, NODE_KIND_WORKER
  from ray.autoscaler._private.constants import DISABLE_LAUNCH_CONFIG_CHECK_KEY, DISABLE_NODE_UPDATERS_KEY, FOREGROUND_NODE_LAUNCH_KEY
label_desc: Use image node_id 3 for calling the BatchingNodeProviderTester obj's underlying member method code with example usage: obj.validate_non_terminated_nodes() without return types