distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for NetDB lookup service."""
2
3import os
4
5import pytest
6
7
8class TestLookupState:
9 def test_enum_values(self):
10 from i2p_netdb.lookup import LookupState
11 assert LookupState.NEW.value == 0
12 assert LookupState.QUERYING.value == 1
13 assert LookupState.FOUND.value == 2
14 assert LookupState.NOT_FOUND.value == 3
15 assert LookupState.FAILED.value == 4
16
17
18class TestLookupRequest:
19 def test_construct(self):
20 from i2p_netdb.lookup import LookupRequest, LookupState
21 key = os.urandom(32)
22 req = LookupRequest(key=key, timeout_ms=5000)
23 assert req.key == key
24 assert req.state == LookupState.NEW
25 assert req.timeout_ms == 5000
26 assert req.result is None
27
28 def test_start(self):
29 from i2p_netdb.lookup import LookupRequest, LookupState
30 req = LookupRequest(key=os.urandom(32))
31 req.start()
32 assert req.state == LookupState.QUERYING
33
34 def test_found(self):
35 from i2p_netdb.lookup import LookupRequest, LookupState
36 req = LookupRequest(key=os.urandom(32))
37 req.start()
38 req.set_found(b"result data")
39 assert req.state == LookupState.FOUND
40 assert req.result == b"result data"
41
42 def test_not_found(self):
43 from i2p_netdb.lookup import LookupRequest, LookupState
44 req = LookupRequest(key=os.urandom(32))
45 req.start()
46 req.set_not_found()
47 assert req.state == LookupState.NOT_FOUND
48
49 def test_failed(self):
50 from i2p_netdb.lookup import LookupRequest, LookupState
51 req = LookupRequest(key=os.urandom(32))
52 req.start()
53 req.set_failed("timeout")
54 assert req.state == LookupState.FAILED
55 assert req.error == "timeout"
56
57 def test_queried_peers(self):
58 from i2p_netdb.lookup import LookupRequest
59 req = LookupRequest(key=os.urandom(32))
60 peer1 = os.urandom(32)
61 peer2 = os.urandom(32)
62 req.add_queried_peer(peer1)
63 req.add_queried_peer(peer2)
64 assert len(req.queried_peers) == 2
65 assert peer1 in req.queried_peers
66
67 def test_exclude_already_queried(self):
68 from i2p_netdb.lookup import LookupRequest
69 req = LookupRequest(key=os.urandom(32))
70 peer = os.urandom(32)
71 req.add_queried_peer(peer)
72 assert req.already_queried(peer)
73 assert not req.already_queried(os.urandom(32))
74
75
76class TestFloodfillSet:
77 def test_add_and_get(self):
78 from i2p_netdb.lookup import FloodfillSet
79 ffs = FloodfillSet()
80 h = os.urandom(32)
81 ffs.add(h, router_info=b"info")
82 assert ffs.is_floodfill(h)
83
84 def test_not_floodfill(self):
85 from i2p_netdb.lookup import FloodfillSet
86 ffs = FloodfillSet()
87 assert not ffs.is_floodfill(os.urandom(32))
88
89 def test_closest_floodfills(self):
90 from i2p_netdb.lookup import FloodfillSet
91 ffs = FloodfillSet()
92 for _ in range(10):
93 ffs.add(os.urandom(32))
94 target = os.urandom(32)
95 closest = ffs.closest(target, n=3)
96 assert len(closest) == 3
97
98 def test_count(self):
99 from i2p_netdb.lookup import FloodfillSet
100 ffs = FloodfillSet()
101 for _ in range(5):
102 ffs.add(os.urandom(32))
103 assert ffs.count() == 5
104
105 def test_remove(self):
106 from i2p_netdb.lookup import FloodfillSet
107 ffs = FloodfillSet()
108 h = os.urandom(32)
109 ffs.add(h)
110 ffs.remove(h)
111 assert not ffs.is_floodfill(h)