Commit afa77741 authored by zhifan huang's avatar zhifan huang

test: add registry.getBootstrap

parent 1aa76589
...@@ -17,9 +17,9 @@ from re6st import registry ...@@ -17,9 +17,9 @@ from re6st import registry
from re6st import ctl from re6st import ctl
from re6st.tests.tools import * from re6st.tests.tools import *
# TODO test for request_dump, requestToken, getNetworkConfig, getBoostrapPeer # TODO test for request_dump, requestToken, getNetworkConfig, getIPv6Address
# getIPV4Information
# getIPV4Information seems useless
def load_config(filename="registry.json"): def load_config(filename="registry.json"):
with open(filename) as f: with open(filename) as f:
...@@ -53,7 +53,10 @@ def select_side_effect(rlist, wlist, elist, timeout): ...@@ -53,7 +53,10 @@ def select_side_effect(rlist, wlist, elist, timeout):
select_side_effect.i = 0 select_side_effect.i = 0
# TODO function for get a unique prefix # TODO function for get a unique prefix
def new_prefix(length=16):
new_prefix.i += 1
return format(new_prefix.i, "0%ib" % length)
new_prefix.i = 0
class TestRegistryServer(unittest.TestCase): class TestRegistryServer(unittest.TestCase):
@classmethod @classmethod
...@@ -337,6 +340,47 @@ class TestRegistryServer(unittest.TestCase): ...@@ -337,6 +340,47 @@ class TestRegistryServer(unittest.TestCase):
self.assertEqual(res, "other msg") self.assertEqual(res, "other msg")
@patch("re6st.registry.RegistryServer._queryAddress")
@patch("re6st.registry.RegistryServer.request_dump", Mock())
def test_getBootstrapPeer(self, query):
"""call request dump situation"""
assert hasattr(self.server, "peers")
peers_back = self.server.peers
self.server.peers = 0, None
cn = new_prefix()
pkey, _ = insert_cert(self.server.db, self.server.cert, cn)
self.server.sessions[cn] = [(0,7)]
query.return_value = expect_msg = "address"
self.server.ctl.neighbours = neigh = {}
# peers while looke like [0, 1, 2, 3, 4], have 5 peer
for i in 1,2,3,4:
neigh[i] = (i, {format(i,"016b"):i})
msg = self.server.getBootstrapPeer(cn)
msg = decrypt(pkey, msg)
peer, msg = msg.split()
self.assertEqual(msg, expect_msg)
self.assertNotIn(peer, self.server.peers)
self.assertEqual(len(self.server.peers[1]), 4)
self.peers = peers_back
@patch("re6st.registry.RegistryServer._queryAddress")
def test_getBoostgetBootstrapPeer_not_dump(self, query):
"""skip the request dump situation, and msg is None"""
assert hasattr(self.server, "peers")
peers_back = self.server.peers
self.server.peers = time.time() + 60, [self.server.prefix]
cn = new_prefix()
query.return_value = None
res = self.server.getBootstrapPeer(cn)
self.assertIsNone(res)
query.assert_called_once_with(self.server.prefix)
self.peers = peers_back
@patch('re6st.registry.RegistryServer.updateNetworkConfig') @patch('re6st.registry.RegistryServer.updateNetworkConfig')
def test_revoke(self, mock_func): def test_revoke(self, mock_func):
# case: no ValueError # case: no ValueError
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment