distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for tunnel building types."""
2
3import os
4
5import pytest
6
7
8class TestBuildRecord:
9 def test_construct(self):
10 from i2p_tunnel.builder import BuildRecord
11 rec = BuildRecord(
12 receive_tunnel_id=1,
13 our_ident=os.urandom(32),
14 next_tunnel_id=2,
15 next_ident=os.urandom(32),
16 layer_key=os.urandom(32),
17 iv_key=os.urandom(32),
18 reply_key=os.urandom(32),
19 reply_iv=os.urandom(16),
20 is_gateway=False,
21 is_endpoint=False,
22 )
23 assert rec.receive_tunnel_id == 1
24 assert rec.next_tunnel_id == 2
25
26 def test_roundtrip(self):
27 from i2p_tunnel.builder import BuildRecord
28 rec = BuildRecord(
29 receive_tunnel_id=100,
30 our_ident=os.urandom(32),
31 next_tunnel_id=200,
32 next_ident=os.urandom(32),
33 layer_key=os.urandom(32),
34 iv_key=os.urandom(32),
35 reply_key=os.urandom(32),
36 reply_iv=os.urandom(16),
37 is_gateway=True,
38 is_endpoint=False,
39 )
40 data = rec.to_bytes()
41 assert len(data) == 222 # Fixed record size
42 rec2 = BuildRecord.from_bytes(data)
43 assert rec2.receive_tunnel_id == 100
44 assert rec2.next_tunnel_id == 200
45 assert rec2.is_gateway
46 assert not rec2.is_endpoint
47
48 def test_endpoint_flag(self):
49 from i2p_tunnel.builder import BuildRecord
50 rec = BuildRecord(
51 receive_tunnel_id=1, our_ident=os.urandom(32),
52 next_tunnel_id=0, next_ident=os.urandom(32),
53 layer_key=os.urandom(32), iv_key=os.urandom(32),
54 reply_key=os.urandom(32), reply_iv=os.urandom(16),
55 is_gateway=False, is_endpoint=True,
56 )
57 data = rec.to_bytes()
58 rec2 = BuildRecord.from_bytes(data)
59 assert rec2.is_endpoint
60 assert not rec2.is_gateway
61
62
63class TestBuildReplyRecord:
64 def test_accepted(self):
65 from i2p_tunnel.builder import BuildReplyRecord
66 rec = BuildReplyRecord(status=0, reply_data=os.urandom(495))
67 assert rec.is_accepted()
68
69 def test_rejected(self):
70 from i2p_tunnel.builder import BuildReplyRecord
71 rec = BuildReplyRecord(status=30, reply_data=os.urandom(495))
72 assert not rec.is_accepted()
73
74 def test_roundtrip(self):
75 from i2p_tunnel.builder import BuildReplyRecord
76 reply_data = os.urandom(495)
77 rec = BuildReplyRecord(status=0, reply_data=reply_data)
78 data = rec.to_bytes()
79 rec2 = BuildReplyRecord.from_bytes(data)
80 assert rec2.status == 0
81 assert rec2.reply_data == reply_data
82
83
84class TestTunnelPool:
85 def test_construct(self):
86 from i2p_tunnel.builder import TunnelPool
87 pool = TunnelPool(name="exploratory")
88 assert pool.name == "exploratory"
89 assert pool.tunnel_count() == 0
90
91 def test_add_tunnel(self):
92 from i2p_tunnel.builder import TunnelPool, TunnelEntry
93 from i2p_data.tunnel import TunnelId
94 pool = TunnelPool(name="test")
95 entry = TunnelEntry(
96 tunnel_id=TunnelId(42),
97 gateway=os.urandom(32),
98 length=3,
99 creation_time=1000,
100 expiration=9999999999999,
101 )
102 pool.add(entry)
103 assert pool.tunnel_count() == 1
104
105 def test_get_tunnel(self):
106 from i2p_tunnel.builder import TunnelPool, TunnelEntry
107 from i2p_data.tunnel import TunnelId
108 pool = TunnelPool(name="test")
109 tid = TunnelId(55)
110 entry = TunnelEntry(tid, os.urandom(32), 2, 1000, 9999999999999)
111 pool.add(entry)
112 found = pool.get(tid)
113 assert found is not None
114 assert int(found.tunnel_id) == 55
115
116 def test_remove_expired(self):
117 from i2p_tunnel.builder import TunnelPool, TunnelEntry
118 from i2p_data.tunnel import TunnelId
119 pool = TunnelPool(name="test")
120 pool.add(TunnelEntry(TunnelId(1), os.urandom(32), 2, 1000, 500)) # expired
121 pool.add(TunnelEntry(TunnelId(2), os.urandom(32), 2, 1000, 9999999999999)) # active
122 pool.remove_expired(now_ms=1000)
123 assert pool.tunnel_count() == 1
124
125 def test_select_random(self):
126 from i2p_tunnel.builder import TunnelPool, TunnelEntry
127 from i2p_data.tunnel import TunnelId
128 pool = TunnelPool(name="test")
129 for i in range(5):
130 pool.add(TunnelEntry(TunnelId(i), os.urandom(32), 2, 1000, 9999999999999))
131 selected = pool.select_random()
132 assert selected is not None
133
134 def test_select_random_empty(self):
135 from i2p_tunnel.builder import TunnelPool
136 pool = TunnelPool(name="test")
137 assert pool.select_random() is None