distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for I2CP message types."""
2
3import os
4import struct
5
6import pytest
7
8
9class TestI2CPMessage:
10 def test_wire_format_length_prefix(self):
11 from i2p_client.i2cp_messages import GetDateMessage, I2CPMessage
12 msg = GetDateMessage(version="0.9.62")
13 wire = msg.to_wire()
14 # Wire format: 4-byte length + 1-byte type + payload
15 length = struct.unpack("!I", wire[:4])[0]
16 assert length == len(wire) - 4 # length does not include itself
17
18 def test_from_wire_dispatches(self):
19 from i2p_client.i2cp_messages import GetDateMessage, I2CPMessage
20 msg = GetDateMessage(version="0.9.62")
21 wire = msg.to_wire()
22 msg2 = I2CPMessage.from_wire(wire)
23 assert isinstance(msg2, GetDateMessage)
24 assert msg2.version == "0.9.62"
25
26 def test_unknown_type(self):
27 from i2p_client.i2cp_messages import I2CPMessage
28 # Build wire with unknown type 254
29 payload = b"test"
30 wire = struct.pack("!IB", len(payload) + 1, 254) + payload
31 with pytest.raises(ValueError, match="Unknown"):
32 I2CPMessage.from_wire(wire)
33
34 def test_from_wire_truncated(self):
35 from i2p_client.i2cp_messages import I2CPMessage
36 with pytest.raises(ValueError):
37 I2CPMessage.from_wire(b"\x00\x00")
38
39
40class TestGetDateMessage:
41 def test_construct(self):
42 from i2p_client.i2cp_messages import GetDateMessage
43 msg = GetDateMessage(version="0.9.62")
44 assert msg.TYPE == 32
45 assert msg.version == "0.9.62"
46
47 def test_roundtrip(self):
48 from i2p_client.i2cp_messages import GetDateMessage, I2CPMessage
49 msg = GetDateMessage(version="0.9.58")
50 wire = msg.to_wire()
51 msg2 = I2CPMessage.from_wire(wire)
52 assert isinstance(msg2, GetDateMessage)
53 assert msg2.version == "0.9.58"
54
55
56class TestSetDateMessage:
57 def test_construct(self):
58 from i2p_client.i2cp_messages import SetDateMessage
59 msg = SetDateMessage(date_ms=1710000000000, version="0.9.62")
60 assert msg.TYPE == 33
61 assert msg.date_ms == 1710000000000
62
63 def test_roundtrip(self):
64 from i2p_client.i2cp_messages import SetDateMessage, I2CPMessage
65 msg = SetDateMessage(date_ms=9999999999999, version="0.9.62")
66 wire = msg.to_wire()
67 msg2 = I2CPMessage.from_wire(wire)
68 assert isinstance(msg2, SetDateMessage)
69 assert msg2.date_ms == 9999999999999
70 assert msg2.version == "0.9.62"
71
72
73class TestSessionStatusMessage:
74 def test_construct(self):
75 from i2p_client.i2cp_messages import SessionStatusMessage
76 msg = SessionStatusMessage(session_id=1, status=0)
77 assert msg.TYPE == 20
78 assert msg.session_id == 1
79 assert msg.status == 0
80
81 def test_status_codes(self):
82 from i2p_client.i2cp_messages import SessionStatusMessage
83 # CREATED=0, UPDATED=1, INVALID=2, REFUSED=3
84 for code in [0, 1, 2, 3]:
85 msg = SessionStatusMessage(session_id=0, status=code)
86 assert msg.status == code
87
88 def test_roundtrip(self):
89 from i2p_client.i2cp_messages import SessionStatusMessage, I2CPMessage
90 msg = SessionStatusMessage(session_id=42, status=1)
91 wire = msg.to_wire()
92 msg2 = I2CPMessage.from_wire(wire)
93 assert isinstance(msg2, SessionStatusMessage)
94 assert msg2.session_id == 42
95 assert msg2.status == 1
96
97
98class TestCreateSessionMessage:
99 def test_roundtrip(self):
100 from i2p_client.i2cp_messages import CreateSessionMessage, I2CPMessage
101 dest_data = os.urandom(387)
102 options = {"inbound.quantity": "3", "outbound.length": "2"}
103 msg = CreateSessionMessage(destination_data=dest_data, options=options)
104 assert msg.TYPE == 1
105 wire = msg.to_wire()
106 msg2 = I2CPMessage.from_wire(wire)
107 assert isinstance(msg2, CreateSessionMessage)
108 assert msg2.destination_data == dest_data
109 assert msg2.options["inbound.quantity"] == "3"
110
111
112class TestDestroySessionMessage:
113 def test_roundtrip(self):
114 from i2p_client.i2cp_messages import DestroySessionMessage, I2CPMessage
115 msg = DestroySessionMessage(session_id=5)
116 assert msg.TYPE == 3
117 wire = msg.to_wire()
118 msg2 = I2CPMessage.from_wire(wire)
119 assert isinstance(msg2, DestroySessionMessage)
120 assert msg2.session_id == 5
121
122
123class TestSendMessageMessage:
124 def test_roundtrip(self):
125 from i2p_client.i2cp_messages import SendMessageMessage, I2CPMessage
126 dest = os.urandom(384) + b'\x00\x00\x00' # 384 key bytes + null cert
127 payload = os.urandom(128)
128 msg = SendMessageMessage(session_id=7, destination_data=dest, payload=payload)
129 assert msg.TYPE == 5
130 wire = msg.to_wire()
131 msg2 = I2CPMessage.from_wire(wire)
132 assert isinstance(msg2, SendMessageMessage)
133 assert msg2.session_id == 7
134 assert msg2.destination_data == dest
135 assert msg2.payload == payload
136
137
138class TestMessageStatusMessage:
139 def test_roundtrip(self):
140 from i2p_client.i2cp_messages import MessageStatusMessage, I2CPMessage
141 msg = MessageStatusMessage(session_id=3, msg_id=9999, nonce=42, status=1, size=100)
142 assert msg.TYPE == 22
143 wire = msg.to_wire()
144 msg2 = I2CPMessage.from_wire(wire)
145 assert isinstance(msg2, MessageStatusMessage)
146 assert msg2.session_id == 3
147 assert msg2.msg_id == 9999
148 assert msg2.nonce == 42
149 assert msg2.status == 1
150 assert msg2.size == 100
151
152
153class TestMessagePayloadMessage:
154 def test_roundtrip(self):
155 from i2p_client.i2cp_messages import MessagePayloadMessage, I2CPMessage
156 payload = os.urandom(256)
157 msg = MessagePayloadMessage(session_id=1, msg_id=555, payload=payload)
158 assert msg.TYPE == 31
159 wire = msg.to_wire()
160 msg2 = I2CPMessage.from_wire(wire)
161 assert isinstance(msg2, MessagePayloadMessage)
162 assert msg2.session_id == 1
163 assert msg2.msg_id == 555
164 assert msg2.payload == payload
165
166
167class TestReceiveMessageBeginMessage:
168 def test_roundtrip(self):
169 from i2p_client.i2cp_messages import ReceiveMessageBeginMessage, I2CPMessage
170 msg = ReceiveMessageBeginMessage(session_id=2, msg_id=100)
171 assert msg.TYPE == 6
172 wire = msg.to_wire()
173 msg2 = I2CPMessage.from_wire(wire)
174 assert isinstance(msg2, ReceiveMessageBeginMessage)
175 assert msg2.session_id == 2
176 assert msg2.msg_id == 100
177
178
179class TestReceiveMessageEndMessage:
180 def test_roundtrip(self):
181 from i2p_client.i2cp_messages import ReceiveMessageEndMessage, I2CPMessage
182 msg = ReceiveMessageEndMessage(session_id=4, msg_id=200)
183 assert msg.TYPE == 7
184 wire = msg.to_wire()
185 msg2 = I2CPMessage.from_wire(wire)
186 assert isinstance(msg2, ReceiveMessageEndMessage)
187 assert msg2.session_id == 4
188 assert msg2.msg_id == 200
189
190
191class TestHostLookupMessage:
192 def test_roundtrip_by_name(self):
193 from i2p_client.i2cp_messages import HostLookupMessage, I2CPMessage
194 msg = HostLookupMessage(session_id=1, request_id=42, hostname="example.i2p")
195 assert msg.TYPE == 38
196 wire = msg.to_wire()
197 msg2 = I2CPMessage.from_wire(wire)
198 assert isinstance(msg2, HostLookupMessage)
199 assert msg2.session_id == 1
200 assert msg2.request_id == 42
201 assert msg2.hostname == "example.i2p"
202
203 def test_roundtrip_by_hash(self):
204 from i2p_client.i2cp_messages import HostLookupMessage, I2CPMessage
205 h = os.urandom(32)
206 msg = HostLookupMessage(session_id=1, request_id=10, dest_hash=h)
207 wire = msg.to_wire()
208 msg2 = I2CPMessage.from_wire(wire)
209 assert isinstance(msg2, HostLookupMessage)
210 assert msg2.dest_hash == h
211 assert msg2.hostname is None
212
213
214class TestHostReplyMessage:
215 def test_roundtrip_success(self):
216 from i2p_client.i2cp_messages import HostReplyMessage, I2CPMessage
217 dest_data = os.urandom(387)
218 msg = HostReplyMessage(session_id=1, request_id=42, result_code=0,
219 destination_data=dest_data)
220 assert msg.TYPE == 39
221 wire = msg.to_wire()
222 msg2 = I2CPMessage.from_wire(wire)
223 assert isinstance(msg2, HostReplyMessage)
224 assert msg2.result_code == 0
225 assert msg2.destination_data == dest_data
226
227 def test_roundtrip_failure(self):
228 from i2p_client.i2cp_messages import HostReplyMessage, I2CPMessage
229 msg = HostReplyMessage(session_id=1, request_id=10, result_code=1)
230 wire = msg.to_wire()
231 msg2 = I2CPMessage.from_wire(wire)
232 assert isinstance(msg2, HostReplyMessage)
233 assert msg2.result_code == 1
234 assert msg2.destination_data is None
235
236
237# --- Tier 0 constant tests ---
238
239def test_session_status_constants():
240 from i2p_client.i2cp_messages import SessionStatusMessage
241 assert SessionStatusMessage.STATUS_DESTROYED == 0
242 assert SessionStatusMessage.STATUS_CREATED == 1
243 assert SessionStatusMessage.STATUS_UPDATED == 2
244 assert SessionStatusMessage.STATUS_INVALID == 3
245 assert SessionStatusMessage.STATUS_REFUSED == 4
246
247def test_message_status_constants():
248 from i2p_client.i2cp_messages import MessageStatusMessage
249 assert MessageStatusMessage.STATUS_AVAILABLE == 0
250 assert MessageStatusMessage.STATUS_SEND_ACCEPTED == 1
251 assert MessageStatusMessage.STATUS_SEND_BEST_EFFORT_SUCCESS == 2
252 assert MessageStatusMessage.STATUS_SEND_BEST_EFFORT_FAILURE == 3
253 assert MessageStatusMessage.STATUS_SEND_GUARANTEED_SUCCESS == 4
254 assert MessageStatusMessage.STATUS_SEND_GUARANTEED_FAILURE == 5
255 assert MessageStatusMessage.STATUS_SEND_SUCCESS_LOCAL == 6
256 assert MessageStatusMessage.STATUS_SEND_FAILURE_LOCAL == 7
257 assert MessageStatusMessage.STATUS_SEND_FAILURE_ROUTER == 8
258 assert MessageStatusMessage.STATUS_SEND_FAILURE_NETWORK == 9
259 assert MessageStatusMessage.STATUS_SEND_FAILURE_BAD_SESSION == 10
260 assert MessageStatusMessage.STATUS_SEND_FAILURE_BAD_MESSAGE == 11
261 assert MessageStatusMessage.STATUS_SEND_FAILURE_BAD_OPTIONS == 12
262 assert MessageStatusMessage.STATUS_SEND_FAILURE_OVERFLOW == 13
263 assert MessageStatusMessage.STATUS_SEND_FAILURE_EXPIRED == 14
264 assert MessageStatusMessage.STATUS_SEND_FAILURE_LOCAL_LEASESET == 15
265 assert MessageStatusMessage.STATUS_SEND_FAILURE_NO_TUNNELS == 16
266 assert MessageStatusMessage.STATUS_SEND_FAILURE_UNSUPPORTED_ENCRYPTION == 17
267 assert MessageStatusMessage.STATUS_SEND_FAILURE_DESTINATION == 18
268 assert MessageStatusMessage.STATUS_SEND_FAILURE_BAD_LEASESET == 19
269 assert MessageStatusMessage.STATUS_SEND_FAILURE_EXPIRED_LEASESET == 20
270 assert MessageStatusMessage.STATUS_SEND_FAILURE_NO_LEASESET == 21
271
272def test_message_status_is_success():
273 from i2p_client.i2cp_messages import MessageStatusMessage
274 for status in (1, 2, 4, 6):
275 msg = MessageStatusMessage(0, 0, 0, status, 0)
276 assert msg.is_success()
277 for status in (0, 3, 5, 7, 8, 9, 10, 21):
278 msg = MessageStatusMessage(0, 0, 0, status, 0)
279 assert not msg.is_success()
280
281def test_message_status_is_failure():
282 from i2p_client.i2cp_messages import MessageStatusMessage
283 for status in (3, 5, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21):
284 msg = MessageStatusMessage(0, 0, 0, status, 0)
285 assert msg.is_failure()
286 for status in (0, 1, 2, 4, 6):
287 msg = MessageStatusMessage(0, 0, 0, status, 0)
288 assert not msg.is_failure()
289
290def test_message_status_name():
291 from i2p_client.i2cp_messages import MessageStatusMessage
292 msg = MessageStatusMessage(0, 0, 0, 0, 0)
293 assert msg.status_name() == "AVAILABLE"
294 msg2 = MessageStatusMessage(0, 0, 0, 4, 0)
295 assert msg2.status_name() == "SEND_GUARANTEED_SUCCESS"
296
297def test_host_reply_result_constants():
298 from i2p_client.i2cp_messages import HostReplyMessage
299 assert HostReplyMessage.RESULT_SUCCESS == 0
300 assert HostReplyMessage.RESULT_FAILURE == 1
301 assert HostReplyMessage.RESULT_SECRET_REQUIRED == 2
302 assert HostReplyMessage.RESULT_KEY_REQUIRED == 3
303 assert HostReplyMessage.RESULT_SECRET_AND_KEY_REQUIRED == 4
304 assert HostReplyMessage.RESULT_DECRYPTION_FAILURE == 5
305
306def test_host_lookup_type_constants():
307 from i2p_client.i2cp_messages import HostLookupMessage
308 assert HostLookupMessage.LOOKUP_HASH == 0
309 assert HostLookupMessage.LOOKUP_HOST == 1
310
311def test_get_date_with_options_roundtrip():
312 from i2p_client.i2cp_messages import GetDateMessage, I2CPMessage
313 msg = GetDateMessage(version="0.9.62", options={"i2cp.username": "alice", "i2cp.password": "secret"})
314 wire = msg.to_wire()
315 msg2 = I2CPMessage.from_wire(wire)
316 assert isinstance(msg2, GetDateMessage)
317 assert msg2.version == "0.9.62"
318 assert msg2.options == {"i2cp.username": "alice", "i2cp.password": "secret"}
319
320def test_get_date_no_options_backward_compat():
321 from i2p_client.i2cp_messages import GetDateMessage, I2CPMessage
322 msg = GetDateMessage(version="0.9.62")
323 wire = msg.to_wire()
324 msg2 = I2CPMessage.from_wire(wire)
325 assert isinstance(msg2, GetDateMessage)
326 assert msg2.version == "0.9.62"
327 assert msg2.options == {}