code: def test_routing_table_search():
      for table_size, lower_active, upper_active in [
          (10, 10, 10),
          (10_000, 800, 1100),
      ]:
          node_id = DHTID.generate()
          routing_table = RoutingTable(
              node_id, bucket_size=20, depth_modulo=5
          )
          num_added = 0
          total_nodes = 0
  
          for phony_neighbor_port in random.sample(
              range(1_000_000), table_size
          ):
              routing_table.add_or_update_node(
                  DHTID.generate(), f"{LOCALHOST}:{phony_neighbor_port}"
              )
              new_total = sum(
                  len(bucket.nodes_to_peer_id)
                  for bucket in routing_table.buckets
              )
              num_added += new_total > total_nodes
              total_nodes = new_total
          num_replacements = sum(
              len(bucket.replacement_nodes)
              for bucket in routing_table.buckets
          )
  
          all_active_neighbors = list(
              chain(
                  *(
                      bucket.nodes_to_peer_id.keys()
                      for bucket in routing_table.buckets
                  )
              )
          )
          assert (
              lower_active <= len(all_active_neighbors) <= upper_active
          )
          assert len(all_active_neighbors) == num_added
          assert num_added + num_replacements == table_size
  
          # random queries
          for i in range(1000):
              k = random.randint(1, 100)
              query_id = DHTID.generate()
              exclude = query_id if random.random() < 0.5 else None
              our_knn, our_peer_ids = zip(
                  *routing_table.get_nearest_neighbors(
                      query_id, k=k, exclude=exclude
                  )
              )
              reference_knn = heapq.nsmallest(
                  k, all_active_neighbors, key=query_id.xor_distance
              )
              assert all(
                  our == ref
                  for our, ref in zip_longest(our_knn, reference_knn)
              )
              assert all(
                  our_peer_id == routing_table[our_node]
                  for our_node, our_peer_id in zip(
                      our_knn, our_peer_ids
                  )
              )
  
          # queries from table
          for i in range(1000):
              k = random.randint(1, 100)
              query_id = random.choice(all_active_neighbors)
              our_knn, our_peer_ids = zip(
                  *routing_table.get_nearest_neighbors(
                      query_id, k=k, exclude=query_id
                  )
              )
  
              reference_knn = heapq.nsmallest(
                  k + 1, all_active_neighbors, key=query_id.xor_distance
              )
              if query_id in reference_knn:
                  reference_knn.remove(query_id)
              assert len(our_knn) == len(reference_knn)
              assert all(
                  query_id.xor_distance(our)
                  == query_id.xor_distance(ref)
                  for our, ref in zip_longest(our_knn, reference_knn)
              )
              assert (
                  routing_table.get_nearest_neighbors(
                      query_id, k=k, exclude=None
                  )[0][0]
                  == query_id
              )
  
file_path: hivemind/tests/test_routing.py
imports: import heapq
  import operator
  import random
  from itertools import chain, zip_longest
  from hivemind.dht.routing import DHTID, RoutingTable
  from hivemind.utils.networking import LOCALHOST
label_desc: Use image node_id 5 for calling a global function with example usage: test_routing_table_search() without return types