distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for TunnelId, HopConfig, and TunnelInfo.
2
3TDD: written before implementation.
4"""
5
6import io
7import os
8import time
9
10import pytest
11
12from i2p_data.tunnel import TunnelId, HopConfig, TunnelInfo
13
14
15# ---------------------------------------------------------------------------
16# TunnelId
17# ---------------------------------------------------------------------------
18
19class TestTunnelId:
20 """TunnelId: 4-byte unsigned integer identifier for tunnels."""
21
22 def test_construction_valid(self):
23 tid = TunnelId(42)
24 assert int(tid) == 42
25
26 def test_construction_zero(self):
27 tid = TunnelId(0)
28 assert int(tid) == 0
29
30 def test_construction_max(self):
31 tid = TunnelId(2**32 - 1)
32 assert int(tid) == 2**32 - 1
33
34 def test_construction_negative_raises(self):
35 with pytest.raises(ValueError):
36 TunnelId(-1)
37
38 def test_construction_too_large_raises(self):
39 with pytest.raises(ValueError):
40 TunnelId(2**32)
41
42 def test_to_bytes(self):
43 tid = TunnelId(0x01020304)
44 assert tid.to_bytes() == b"\x01\x02\x03\x04"
45
46 def test_to_bytes_zero(self):
47 tid = TunnelId(0)
48 assert tid.to_bytes() == b"\x00\x00\x00\x00"
49
50 def test_from_bytes_roundtrip(self):
51 original = TunnelId(12345)
52 restored = TunnelId.from_bytes(original.to_bytes())
53 assert original == restored
54
55 def test_from_bytes_specific(self):
56 tid = TunnelId.from_bytes(b"\x00\x00\x00\x2a")
57 assert int(tid) == 42
58
59 def test_from_stream(self):
60 stream = io.BytesIO(b"\x00\x00\x01\x00extra")
61 tid = TunnelId.from_stream(stream)
62 assert int(tid) == 256
63 # Stream should have advanced past the 4 bytes
64 assert stream.read() == b"extra"
65
66 def test_equality(self):
67 assert TunnelId(100) == TunnelId(100)
68
69 def test_inequality(self):
70 assert TunnelId(100) != TunnelId(200)
71
72 def test_equality_different_type(self):
73 assert TunnelId(100) != 100
74
75 def test_hash_same(self):
76 assert hash(TunnelId(100)) == hash(TunnelId(100))
77
78 def test_hash_usable_in_set(self):
79 s = {TunnelId(1), TunnelId(2), TunnelId(1)}
80 assert len(s) == 2
81
82 def test_is_zero_true(self):
83 assert TunnelId(0).is_zero() is True
84
85 def test_is_zero_false(self):
86 assert TunnelId(1).is_zero() is False
87
88 def test_repr(self):
89 r = repr(TunnelId(42))
90 assert "42" in r
91
92 def test_int(self):
93 tid = TunnelId(999)
94 assert int(tid) == 999
95
96
97# ---------------------------------------------------------------------------
98# HopConfig
99# ---------------------------------------------------------------------------
100
101class TestHopConfig:
102 """HopConfig: configuration for one hop in a tunnel."""
103
104 def _make_hop_config(self):
105 return HopConfig(
106 receive_tunnel_id=TunnelId(1),
107 send_tunnel_id=TunnelId(2),
108 receive_key=os.urandom(32),
109 send_key=os.urandom(32),
110 iv_key=os.urandom(32),
111 reply_key=os.urandom(32),
112 reply_iv=os.urandom(16),
113 layer_key=os.urandom(32),
114 )
115
116 def test_construction(self):
117 hc = self._make_hop_config()
118 assert isinstance(hc.receive_tunnel_id, TunnelId)
119 assert isinstance(hc.send_tunnel_id, TunnelId)
120
121 def test_field_access(self):
122 rk = os.urandom(32)
123 sk = os.urandom(32)
124 ivk = os.urandom(32)
125 rpk = os.urandom(32)
126 rpiv = os.urandom(16)
127 lk = os.urandom(32)
128 hc = HopConfig(
129 receive_tunnel_id=TunnelId(10),
130 send_tunnel_id=TunnelId(20),
131 receive_key=rk,
132 send_key=sk,
133 iv_key=ivk,
134 reply_key=rpk,
135 reply_iv=rpiv,
136 layer_key=lk,
137 )
138 assert int(hc.receive_tunnel_id) == 10
139 assert int(hc.send_tunnel_id) == 20
140 assert hc.receive_key == rk
141 assert hc.send_key == sk
142 assert hc.iv_key == ivk
143 assert hc.reply_key == rpk
144 assert hc.reply_iv == rpiv
145 assert hc.layer_key == lk
146
147
148# ---------------------------------------------------------------------------
149# TunnelInfo
150# ---------------------------------------------------------------------------
151
152class TestTunnelInfo:
153 """TunnelInfo: metadata about a built tunnel."""
154
155 def _make_tunnel_info(self, **overrides):
156 defaults = dict(
157 tunnel_id=TunnelId(500),
158 gateway=os.urandom(32),
159 length=3,
160 creation_time=1_000_000,
161 expiration=2_000_000,
162 )
163 defaults.update(overrides)
164 return TunnelInfo(**defaults)
165
166 def test_construction(self):
167 ti = self._make_tunnel_info()
168 assert int(ti.tunnel_id) == 500
169 assert ti.length == 3
170 assert ti.creation_time == 1_000_000
171 assert ti.expiration == 2_000_000
172
173 def test_gateway_size(self):
174 gw = os.urandom(32)
175 ti = self._make_tunnel_info(gateway=gw)
176 assert ti.gateway == gw
177 assert len(ti.gateway) == 32
178
179 def test_gateway_wrong_size_raises(self):
180 with pytest.raises(ValueError):
181 self._make_tunnel_info(gateway=b"\x00" * 16)
182
183 def test_is_expired_past(self):
184 ti = self._make_tunnel_info(expiration=1_000)
185 assert ti.is_expired(now_ms=2_000) is True
186
187 def test_is_expired_future(self):
188 ti = self._make_tunnel_info(expiration=5_000)
189 assert ti.is_expired(now_ms=1_000) is False
190
191 def test_is_expired_exact(self):
192 ti = self._make_tunnel_info(expiration=1_000)
193 # At exact expiration time, should be expired (<=)
194 assert ti.is_expired(now_ms=1_000) is True
195
196 def test_is_expired_default_now(self):
197 # Far-future expiration should not be expired
198 ti = self._make_tunnel_info(expiration=int(time.time() * 1000) + 600_000)
199 assert ti.is_expired() is False
200
201 def test_equality(self):
202 gw = os.urandom(32)
203 a = self._make_tunnel_info(tunnel_id=TunnelId(1), gateway=gw)
204 b = self._make_tunnel_info(tunnel_id=TunnelId(1), gateway=gw)
205 assert a == b
206
207 def test_inequality(self):
208 gw = os.urandom(32)
209 a = self._make_tunnel_info(tunnel_id=TunnelId(1), gateway=gw)
210 b = self._make_tunnel_info(tunnel_id=TunnelId(2), gateway=gw)
211 assert a != b
212
213 def test_equality_different_type(self):
214 ti = self._make_tunnel_info()
215 assert ti != "not a tunnel"
216
217 def test_repr(self):
218 r = repr(self._make_tunnel_info())
219 assert "500" in r