distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for I2NP tunnel message types."""
2
3import os
4import struct
5
6import pytest
7
8
9class TestTunnelDataMessage:
10 def test_construct(self):
11 from i2p_data.i2np_tunnel import TunnelDataMessage
12 data = os.urandom(1024)
13 msg = TunnelDataMessage(tunnel_id=12345, data=data)
14 assert msg.tunnel_id == 12345
15 assert msg.data == data
16 assert msg.TYPE == 18
17
18 def test_body_size(self):
19 from i2p_data.i2np_tunnel import TunnelDataMessage
20 msg = TunnelDataMessage(tunnel_id=1, data=os.urandom(1024))
21 assert len(msg.body_bytes()) == 4 + 1024
22
23 def test_roundtrip(self):
24 from i2p_data.i2np import I2NPMessage
25 from i2p_data.i2np_tunnel import TunnelDataMessage
26 data = os.urandom(1024)
27 msg = TunnelDataMessage(tunnel_id=0xDEAD, data=data)
28 wire = msg.to_bytes()
29 msg2 = I2NPMessage.from_bytes(wire)
30 assert isinstance(msg2, TunnelDataMessage)
31 assert msg2.tunnel_id == 0xDEAD
32 assert msg2.data == data
33
34
35class TestTunnelGatewayMessage:
36 def test_construct(self):
37 from i2p_data.i2np_tunnel import TunnelGatewayMessage
38 payload = b"gateway payload"
39 msg = TunnelGatewayMessage(tunnel_id=999, data=payload)
40 assert msg.tunnel_id == 999
41 assert msg.data == payload
42 assert msg.TYPE == 19
43
44 def test_body_format(self):
45 from i2p_data.i2np_tunnel import TunnelGatewayMessage
46 payload = b"test"
47 msg = TunnelGatewayMessage(tunnel_id=1, data=payload)
48 body = msg.body_bytes()
49 tid = struct.unpack("!I", body[:4])[0]
50 length = struct.unpack("!H", body[4:6])[0]
51 assert tid == 1
52 assert length == 4
53 assert body[6:] == payload
54
55 def test_roundtrip(self):
56 from i2p_data.i2np import I2NPMessage
57 from i2p_data.i2np_tunnel import TunnelGatewayMessage
58 payload = os.urandom(200)
59 msg = TunnelGatewayMessage(tunnel_id=5555, data=payload)
60 wire = msg.to_bytes()
61 msg2 = I2NPMessage.from_bytes(wire)
62 assert isinstance(msg2, TunnelGatewayMessage)
63 assert msg2.tunnel_id == 5555
64 assert msg2.data == payload
65
66
67class TestTunnelBuildMessage:
68 def test_construct(self):
69 from i2p_data.i2np_tunnel import TunnelBuildMessage
70 records = [os.urandom(528) for _ in range(8)]
71 msg = TunnelBuildMessage(records=records)
72 assert msg.TYPE == 21
73 assert len(msg.records) == 8
74
75 def test_record_count_validation(self):
76 from i2p_data.i2np_tunnel import TunnelBuildMessage
77 with pytest.raises(ValueError):
78 TunnelBuildMessage(records=[os.urandom(528) for _ in range(7)])
79
80 def test_body_size(self):
81 from i2p_data.i2np_tunnel import TunnelBuildMessage
82 msg = TunnelBuildMessage(records=[os.urandom(528) for _ in range(8)])
83 assert len(msg.body_bytes()) == 8 * 528
84
85 def test_roundtrip(self):
86 from i2p_data.i2np import I2NPMessage
87 from i2p_data.i2np_tunnel import TunnelBuildMessage
88 records = [os.urandom(528) for _ in range(8)]
89 msg = TunnelBuildMessage(records=records)
90 wire = msg.to_bytes()
91 msg2 = I2NPMessage.from_bytes(wire)
92 assert isinstance(msg2, TunnelBuildMessage)
93 for i in range(8):
94 assert msg2.records[i] == records[i]
95
96
97class TestTunnelBuildReplyMessage:
98 def test_construct(self):
99 from i2p_data.i2np_tunnel import TunnelBuildReplyMessage
100 records = [os.urandom(528) for _ in range(8)]
101 msg = TunnelBuildReplyMessage(records=records)
102 assert msg.TYPE == 22
103
104 def test_roundtrip(self):
105 from i2p_data.i2np import I2NPMessage
106 from i2p_data.i2np_tunnel import TunnelBuildReplyMessage
107 records = [os.urandom(528) for _ in range(8)]
108 msg = TunnelBuildReplyMessage(records=records)
109 wire = msg.to_bytes()
110 msg2 = I2NPMessage.from_bytes(wire)
111 assert isinstance(msg2, TunnelBuildReplyMessage)
112 assert len(msg2.records) == 8
113
114
115class TestVariableTunnelBuildMessage:
116 def test_construct(self):
117 from i2p_data.i2np_tunnel import VariableTunnelBuildMessage
118 records = [os.urandom(528) for _ in range(3)]
119 msg = VariableTunnelBuildMessage(records=records)
120 assert msg.TYPE == 23
121 assert len(msg.records) == 3
122
123 def test_roundtrip_various_counts(self):
124 from i2p_data.i2np import I2NPMessage
125 from i2p_data.i2np_tunnel import VariableTunnelBuildMessage
126 for n in [1, 4, 8]:
127 records = [os.urandom(528) for _ in range(n)]
128 msg = VariableTunnelBuildMessage(records=records)
129 wire = msg.to_bytes()
130 msg2 = I2NPMessage.from_bytes(wire)
131 assert isinstance(msg2, VariableTunnelBuildMessage)
132 assert len(msg2.records) == n
133
134 def test_body_format(self):
135 from i2p_data.i2np_tunnel import VariableTunnelBuildMessage
136 records = [os.urandom(528) for _ in range(5)]
137 msg = VariableTunnelBuildMessage(records=records)
138 body = msg.body_bytes()
139 assert body[0] == 5 # record count byte
140 assert len(body) == 1 + 5 * 528
141
142
143class TestVariableTunnelBuildReplyMessage:
144 def test_construct(self):
145 from i2p_data.i2np_tunnel import VariableTunnelBuildReplyMessage
146 records = [os.urandom(528) for _ in range(2)]
147 msg = VariableTunnelBuildReplyMessage(records=records)
148 assert msg.TYPE == 24
149
150 def test_roundtrip(self):
151 from i2p_data.i2np import I2NPMessage
152 from i2p_data.i2np_tunnel import VariableTunnelBuildReplyMessage
153 records = [os.urandom(528) for _ in range(6)]
154 msg = VariableTunnelBuildReplyMessage(records=records)
155 wire = msg.to_bytes()
156 msg2 = I2NPMessage.from_bytes(wire)
157 assert isinstance(msg2, VariableTunnelBuildReplyMessage)
158 assert len(msg2.records) == 6
159 for i in range(6):
160 assert msg2.records[i] == records[i]