code: def test_sort_local_workers_by_gpu_id(ray_start_2_cpus):
      def create_worker_group(pids, ips, gpu_ids):
          wg = WorkerGroup(num_workers=2)
          wg.workers = [
              Worker(
                  actor=None,
                  metadata=WorkerMetadata(
                      node_id="dummy",
                      node_ip=ip,
                      hostname="dummy",
                      resource_ids={
                          "GPU": gpu_id.split() if gpu_id else []
                      },
                      pid=pid,
                  ),
              )
              for pid, ip, gpu_id in zip(pids, ips, gpu_ids)
          ]
          return wg
  
      def setup_and_check_worker_group(
          pids, ips, gpu_ids, expected_local_ranks
      ):
          """
          Create a worker group, group workers by IP, and check local ranks assignment.
  
          Args:
              pids: List of unique process IDs.
              ips: List of IP addresses corresponding to each PID.
              gpu_ids: List of GPU IDs or None for each PID.
              expected_local_ranks: Dictionary mapping PID to the
                  expected local rank.
          """
          wg = create_worker_group(pids=pids, ips=ips, gpu_ids=gpu_ids)
          wg.sort_workers_by_ip_and_gpu_id()
  
          # Build local ranks according to the logics in
          # `BackendExecutor._create_rank_world_size_mappings()`
          ip_dict = defaultdict(int)
          local_ranks_map = defaultdict(int)
          for w in wg.workers:
              local_ranks_map[w.metadata.pid] = ip_dict[
                  w.metadata.node_ip
              ]
              ip_dict[w.metadata.node_ip] += 1
  
          local_ranks = [local_ranks_map[pid] for pid in pids]
  
          assert (
              local_ranks == expected_local_ranks
          ), "Incorrect local ranks allocation!\n"
          f"Expect: {expected_local_ranks}\nGot: {local_ranks}"
  
      # Define the worker configurations for different scenarios
      # For workers without GPU resources, their original order will be preserved
      cpu_workers_config = {
          "pids": [0, 1, 2, 3, 4, 5, 6, 7],
          "ips": ["2", "2", "1", "1", "2", "1", "1", "2"],
          "gpu_ids": [None] * 8,
          "expected_local_ranks": [0, 1, 0, 1, 2, 2, 3, 3],
      }
  
      gpu_workers_single_gpu_config = {
          "pids": [0, 1, 2, 3, 4, 5, 6, 7],
          "ips": ["2", "2", "1", "1", "2", "1", "1", "2"],
          "gpu_ids": ["1", "0", "3", "2", "2", "0", "1", "3"],
          "expected_local_ranks": [1, 0, 3, 2, 2, 0, 1, 3],
      }
  
      # For workers with multiple gpus, sort by their lowest gpu id
      gpu_workers_multiple_gpus_config = {
          "pids": [0, 1, 2, 3],
          "ips": ["2", "1", "1", "2"],
          "gpu_ids": ["1,3", "2,1", "0,3", "0,2"],
          "expected_local_ranks": [1, 1, 0, 0],
      }
  
      # Setup and check worker groups for each configuration
      setup_and_check_worker_group(**cpu_workers_config)
      setup_and_check_worker_group(**gpu_workers_single_gpu_config)
      setup_and_check_worker_group(**gpu_workers_multiple_gpus_config)
  
imports: import time
  from collections import defaultdict
  import pytest
  import ray
  import ray._private.ray_constants
  from ray.train._internal.worker_group import Worker, WorkerGroup, WorkerMetadata
label_desc: Use image node_id 14 for calling a global function with example usage: test_sort_local_workers_by_gpu_id(ray_start_2_cpus) and returns: wg