code: def test_locality_manager_busyness_ranking(self):
      pool = _ActorPool(max_tasks_in_flight=2)
  
      # Setup bundle mocks.
      bundles = make_ref_bundles([[0] for _ in range(10)])
      fake_loc_map = {}
      # Also test unknown location handling.
      for i, b in enumerate(bundles):
          fake_loc_map[b] = None
      pool._get_location = lambda b: fake_loc_map[b]
  
      # Setup two actors on the same node.
      actor1 = PoolWorker.remote(node_id="node1")
      ready_ref = actor1.get_location.remote()
      pool.add_pending_actor(actor1, ready_ref)
      ray.get(ready_ref)
      pool.pending_to_running(ready_ref)
      actor2 = PoolWorker.remote(node_id="node1")
      ready_ref = actor2.get_location.remote()
      pool.add_pending_actor(actor2, ready_ref)
      ray.get(ready_ref)
      pool.pending_to_running(ready_ref)
  
      # Fake actor 2 as more busy.
      pool._num_tasks_in_flight[actor2] = 1
      res1 = pool.pick_actor(bundles[0])
      assert res1 == actor1
  
      # Fake actor 2 as more busy again.
      pool._num_tasks_in_flight[actor2] = 2
      res2 = pool.pick_actor(bundles[0])
      assert res2 == actor1
  
      # Nothing left
      res3 = pool.pick_actor(bundles[0])
      assert res3 is None
  
imports: import collections
  import time
  import pytest
  import ray
  from ray.data._internal.compute import ActorPoolStrategy
  from ray.data._internal.execution.operators.actor_pool_map_operator import AutoscalingConfig, AutoscalingPolicy, _ActorPool
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.tests.conftest import *
label_desc: Use image node_id 22 for calling the TestActorPool obj's underlying member method code with example usage: obj.test_locality_manager_busyness_ranking() without return types