6
6
from libp2p .crypto .secp256k1 import (
7
7
create_new_key_pair ,
8
8
)
9
+ from libp2p .peer .id import ID
9
10
from libp2p .peer .peerdata import (
10
11
PeerData ,
11
12
PeerDataError ,
12
13
)
14
+ from libp2p .peer .peerstore import PeerStore
13
15
14
16
MOCK_ADDR = Multiaddr ("/ip4/127.0.0.1/tcp/4001" )
15
17
MOCK_KEYPAIR = create_new_key_pair ()
@@ -60,6 +62,7 @@ def test_supports_protocols():
60
62
assert supported == ["protocol1" , "protocol2" ]
61
63
62
64
65
+ # Test case for first supported protocol is found
63
66
def test_first_supported_protocol_found ():
64
67
peer_data = PeerData ()
65
68
peer_data .set_protocols (["protocolA" , "protocolB" ])
@@ -70,6 +73,7 @@ def test_first_supported_protocol_found():
70
73
assert first == "protocolB"
71
74
72
75
76
+ # Test case for first supported protocol not found
73
77
def test_first_supported_protocol_none ():
74
78
peer_data = PeerData ()
75
79
peer_data .set_protocols (["protocolX" , "protocolY" ])
@@ -80,6 +84,7 @@ def test_first_supported_protocol_none():
80
84
assert first == "None supported"
81
85
82
86
87
+ # Test case for clearing protocol data
83
88
def test_clear_protocol_data ():
84
89
peer_data = PeerData ()
85
90
peer_data .set_protocols (["proto1" , "proto2" ])
@@ -159,6 +164,42 @@ def test_get_privkey_not_found():
159
164
peer_data .get_privkey ()
160
165
161
166
167
+ # Test case for returning all the peers with stored keys
168
+ def test_peer_with_keys ():
169
+ peer_store = PeerStore ()
170
+ peer_id_1 = ID (b"peer1" )
171
+ peer_id_2 = ID (b"peer2" )
172
+
173
+ peer_data_1 = PeerData ()
174
+ peer_data_2 = PeerData ()
175
+
176
+ peer_data_1 .pubkey = MOCK_PUBKEY
177
+ peer_data_2 .pubkey = None
178
+
179
+ peer_store .peer_data_map = {
180
+ peer_id_1 : peer_data_1 ,
181
+ peer_id_2 : peer_data_2 ,
182
+ }
183
+
184
+ assert peer_store .peer_with_keys () == [peer_id_1 ]
185
+
186
+
187
+ # Test case for clearing the key book
188
+ def test_clear_keydata ():
189
+ peer_store = PeerStore ()
190
+ peer_id = ID (b"peer123" )
191
+ peer_data = PeerData ()
192
+
193
+ peer_data .pubkey = MOCK_PUBKEY
194
+ peer_data .privkey = MOCK_PRIVKEY
195
+ peer_store .peer_data_map = {peer_id : peer_data }
196
+
197
+ peer_store .clear_keydata (peer_id )
198
+
199
+ assert peer_data .pubkey is None
200
+ assert peer_data .privkey is None
201
+
202
+
162
203
# Test case for recording latency for the first time
163
204
def test_record_latency_initial ():
164
205
peer_data = PeerData ()
0 commit comments