code: def test_locality_manager_actor_ranking(self):
      pool = _ActorPool(max_tasks_in_flight=2)
  
      # Setup bundle mocks.
      bundles = make_ref_bundles([[0] for _ in range(10)])
      fake_loc_map = {}
      for i, b in enumerate(bundles):
          fake_loc_map[b] = "node1"
      pool._get_location = lambda b: fake_loc_map[b]
  
      # Setup an actor on each 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="node2")
      ready_ref = actor2.get_location.remote()
      pool.add_pending_actor(actor2, ready_ref)
      ray.get(ready_ref)
      pool.pending_to_running(ready_ref)
  
      # Actors on node1 should be preferred.
      res1 = pool.pick_actor(bundles[0])
      assert res1 == actor1
      res2 = pool.pick_actor(bundles[1])
      assert res2 == actor1
  
      # Fallback to remote actors.
      res3 = pool.pick_actor(bundles[2])
      assert res3 == actor2
      res4 = pool.pick_actor(bundles[3])
      assert res4 == actor2
      res5 = pool.pick_actor(bundles[4])
      assert res5 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 21 for calling the TestActorPool obj's underlying member method code with example usage: obj.test_locality_manager_actor_ranking() without return types