code: def test_dht_single_node():
      node = hivemind.DHT(start=True)
      beam_search = MoEBeamSearcher(node, "expert.", grid_size=(10,))
  
      assert all(
          declare_experts(
              node,
              ["expert.1", "expert.2", "expert.3"],
              get_dht_time() + 30,
          ).values()
      )
      assert (
          len(
              declare_experts(
                  node, ["ffn.1", "ffn.2"], get_dht_time() + 30
              )
          )
          == 4
      )
      assert (
          len(
              declare_experts(
                  node,
                  ["e.1.2.3", "e.1.2.5", "e.2.0"],
                  get_dht_time() + 30,
              )
          )
          == 7
      )
  
      for expert in get_experts(node, ["expert.3", "expert.2"]):
          assert expert.peer_id == node.peer_id
  
      assert all(
          declare_experts(
              node, ["expert.5", "expert.2"], get_dht_time() + 30
          ).values()
      )
      found_experts = beam_search.find_best_experts(
          [(0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0)], beam_size=2
      )
      assert len(found_experts) == 2 and [
          expert.uid for expert in found_experts
      ] == ["expert.5", "expert.3"]
  
      successors = beam_search.get_active_successors(
          ["e.1.2.", "e.2.", "e.4.5."]
      )
      assert len(successors["e.1.2."]) == 2
  
      assert successors["e.1.2."][3] == ExpertInfo(
          "e.1.2.3", node.peer_id
      )
      assert successors["e.1.2."][5] == ExpertInfo(
          "e.1.2.5", node.peer_id
      )
      assert len(successors["e.2."]) == 1 and successors["e.2."][
          0
      ] == ExpertInfo("e.2.0", node.peer_id)
      assert successors["e.4.5."] == {}
  
      initial_beam = beam_search.get_initial_beam(
          (3, 2, 1, 0, -1, -2, -3), beam_size=3
      )
      assert len(initial_beam) == 3
      assert initial_beam[0][:2] == (2.0, "expert.1.")
      assert initial_beam[1][:2] == (1.0, "expert.2.")
      assert initial_beam[2][:2] == (0.0, "expert.3.")
  
      with pytest.raises(AssertionError):
          beam_search = MoEBeamSearcher(node, "expert.1.ffn", (2, 2))
  
      with pytest.raises(AssertionError):
          beam_search.get_active_successors(["e.1.2.", "e.2", "e.4.5."])
  
file_path: hivemind/tests/test_dht_experts.py
imports: import asyncio
  import random
  import time
  import numpy
  import pytest
  import hivemind
  from hivemind import get_dht_time
  from hivemind.dht.node import DHTNode
  from hivemind.moe.client.beam_search import MoEBeamSearcher
  from hivemind.moe.expert_uid import ExpertInfo, is_valid_prefix, is_valid_uid, split_uid
  from hivemind.moe.server.dht_handler import declare_experts, get_experts
label_desc: Use image node_id 3 for calling a global function with example usage: test_dht_single_node() without return types