code: def test_counting_resources(start_connected_cluster, storage):
      """Tests that Tune accounting is consistent with actual cluster."""
  
      cluster = start_connected_cluster
      nodes = []
      assert ray.cluster_resources()["CPU"] == 1
      runner = TuneController(
          search_alg=BasicVariantGenerator(), storage=storage
      )
      kwargs = {
          "stopping_criterion": {"training_iteration": 10},
          "config": {"sleep": 1.5},
          "storage": storage,
      }
  
      trials = [Trial("__fake", **kwargs), Trial("__fake", **kwargs)]
      for t in trials:
          runner.add_trial(t)
  
      while not any(t.status == Trial.RUNNING for t in trials):
          runner.step()
  
      running_trials = _get_running_trials(runner)
      assert len(running_trials) == 1
      assert _check_trial_running(running_trials[0])
      assert ray.available_resources().get("CPU", 0) == 0
      nodes += [cluster.add_node(num_cpus=1)]
      cluster.wait_for_nodes()
      assert ray.cluster_resources()["CPU"] == 2
      cluster.remove_node(nodes.pop())
      cluster.wait_for_nodes()
      assert ray.cluster_resources()["CPU"] == 1
  
      while not any(t.status == Trial.RUNNING for t in trials):
          runner.step()
  
      # Only 1 trial can be running due to resource limitation.
      assert (
          sum(t.status == Trial.RUNNING for t in runner.get_trials())
          == 1
      )
  
      for i in range(5):
          nodes += [cluster.add_node(num_cpus=1)]
      cluster.wait_for_nodes()
      assert ray.cluster_resources()["CPU"] == 6
  
      while any(t.status == Trial.PENDING for t in trials):
          runner.step()
      assert (
          sum(t.status == Trial.RUNNING for t in runner.get_trials())
          == 2
      ), [t.status for t in trials]
  
imports: import time
  import os
  import pytest
  import sys
  from unittest.mock import MagicMock
  import ray
  from ray import tune
  from ray.train import CheckpointConfig
  from ray.cluster_utils import Cluster
  from ray.train._internal.storage import StorageContext
  from ray.tune.error import TuneError
  from ray.tune.search import BasicVariantGenerator
  from ray.tune.experiment import Trial
  from ray.tune.execution.tune_controller import TuneController
label_desc: Use image node_id 7 for calling a global function with example usage: test_counting_resources(start_connected_cluster, storage) without return types