code: def test_min_worker_nodes():
      scheduler = ResourceDemandScheduler()
      node_type_configs = {
          "type_1": NodeTypeConfig(
              name="type_1",
              resources={"CPU": 1},
              min_worker_nodes=1,
              max_worker_nodes=10,
          ),
          "type_2": NodeTypeConfig(
              name="type_2",
              resources={"CPU": 1},
              min_worker_nodes=0,
              max_worker_nodes=10,
          ),
          "type_3": NodeTypeConfig(
              name="type_3",
              resources={"CPU": 1},
              min_worker_nodes=2,
              max_worker_nodes=10,
          ),
      }
      # With empty cluster
      request = sched_request(
          node_type_configs=node_type_configs,
      )
  
      reply = scheduler.schedule(request)
  
      target_cluster_shape = {"type_1": 1, "type_3": 2}
      assert sorted(reply.target_cluster_shape) == sorted(
          target_cluster_shape
      )
  
      # With existing ray nodes
      request = sched_request(
          node_type_configs=node_type_configs,
          current_nodes=[
              NodeState(ray_node_type_name="type_1"),
              NodeState(ray_node_type_name="type_1"),
          ],
      )
  
      reply = scheduler.schedule(request)
      target_cluster_shape = {"type_1": 2, "type_3": 2}
      assert sorted(reply.target_cluster_shape) == sorted(
          target_cluster_shape
      )
  
      # With existing instances pending.
      request = sched_request(
          node_type_configs=node_type_configs,
          current_instances=[
              Instance(
                  instance_type="type_1", status=Instance.REQUESTED
              ),
              Instance(
                  instance_type="type_1", status=Instance.ALLOCATED
              ),
              Instance(
                  instance_type="type_no_longer_exists",
                  status=Instance.REQUESTED,
              ),
          ],
      )
      target_cluster_shape = {"type_1": 2, "type_3": 2}
      reply = scheduler.schedule(request)
      assert sorted(reply.target_cluster_shape) == sorted(
          target_cluster_shape
      )
  
file_path: ray/python/ray/autoscaler/v2/tests/test_scheduler.py
imports: import os
  import sys
  from typing import Dict, List, Optional
  import pytest
  from ray.autoscaler.v2.scheduler import NodeTypeConfig, ResourceDemandScheduler, SchedulingRequest
  from ray.autoscaler.v2.schema import NodeType
  from ray.core.generated.autoscaler_pb2 import ClusterResourceConstraint, GangResourceRequest, NodeState, ResourceRequestByCount
  from ray.core.generated.instance_manager_pb2 import Instance
label_desc: Use image node_id 2 for calling a global function with example usage: test_min_worker_nodes() without return types