distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for I2CP Tier 1 message types."""
2
3import struct
4import pytest
5
6from i2p_client.i2cp_messages import (
7 I2CPMessage,
8 DisconnectMessage,
9 GetBandwidthLimitsMessage,
10 BandwidthLimitsMessage,
11 DestLookupMessage,
12 DestReplyMessage,
13 ReportAbuseMessage,
14)
15
16
17class TestDisconnectMessage:
18 def test_roundtrip(self):
19 msg = DisconnectMessage("session expired")
20 wire = msg.to_wire()
21 msg2 = I2CPMessage.from_wire(wire)
22 assert isinstance(msg2, DisconnectMessage)
23 assert msg2.reason == "session expired"
24
25 def test_empty_reason(self):
26 msg = DisconnectMessage("")
27 wire = msg.to_wire()
28 msg2 = I2CPMessage.from_wire(wire)
29 assert msg2.reason == ""
30
31
32class TestGetBandwidthLimitsMessage:
33 def test_roundtrip(self):
34 msg = GetBandwidthLimitsMessage()
35 wire = msg.to_wire()
36 msg2 = I2CPMessage.from_wire(wire)
37 assert isinstance(msg2, GetBandwidthLimitsMessage)
38
39 def test_empty_payload(self):
40 msg = GetBandwidthLimitsMessage()
41 assert msg.payload_bytes() == b""
42
43
44class TestBandwidthLimitsMessage:
45 def test_roundtrip(self):
46 limits = list(range(16))
47 msg = BandwidthLimitsMessage(limits)
48 wire = msg.to_wire()
49 msg2 = I2CPMessage.from_wire(wire)
50 assert isinstance(msg2, BandwidthLimitsMessage)
51 assert msg2.limits == limits
52
53 def test_named_accessors(self):
54 limits = [100, 200, 300, 0, 400] + [0] * 11
55 msg = BandwidthLimitsMessage(limits)
56 assert msg.client_inbound == 100
57 assert msg.client_outbound == 200
58 assert msg.router_inbound == 300
59 assert msg.router_outbound == 400
60
61 def test_payload_size(self):
62 msg = BandwidthLimitsMessage([0] * 16)
63 assert len(msg.payload_bytes()) == 64
64
65 def test_wrong_count_raises(self):
66 with pytest.raises(ValueError):
67 BandwidthLimitsMessage([1, 2, 3])
68
69
70class TestDestLookupMessage:
71 def test_roundtrip(self):
72 h = b"\xab" * 32
73 msg = DestLookupMessage(h)
74 wire = msg.to_wire()
75 msg2 = I2CPMessage.from_wire(wire)
76 assert isinstance(msg2, DestLookupMessage)
77 assert msg2.dest_hash == h
78
79 def test_wrong_size_raises(self):
80 with pytest.raises(ValueError):
81 DestLookupMessage(b"\x00" * 16)
82
83
84class TestDestReplyMessage:
85 def test_success(self):
86 dest_data = b"\x01" * 387
87 msg = DestReplyMessage(destination_data=dest_data)
88 wire = msg.to_wire()
89 msg2 = I2CPMessage.from_wire(wire)
90 assert msg2.destination_data == dest_data
91
92 def test_hash_failure(self):
93 h = b"\x02" * 32
94 msg = DestReplyMessage(hash_data=h)
95 wire = msg.to_wire()
96 msg2 = I2CPMessage.from_wire(wire)
97 assert msg2.hash_data == h
98
99 def test_empty_failure(self):
100 msg = DestReplyMessage()
101 wire = msg.to_wire()
102 msg2 = I2CPMessage.from_wire(wire)
103 assert msg2.destination_data is None
104 assert msg2.hash_data is None
105
106
107class TestReportAbuseMessage:
108 def test_roundtrip(self):
109 msg = ReportAbuseMessage(session_id=5, severity=3, reason="spam", message_id=999)
110 wire = msg.to_wire()
111 msg2 = I2CPMessage.from_wire(wire)
112 assert isinstance(msg2, ReportAbuseMessage)
113 assert msg2.session_id == 5
114 assert msg2.severity == 3
115 assert msg2.reason == "spam"
116 assert msg2.message_id == 999
117
118 def test_empty_reason(self):
119 msg = ReportAbuseMessage(0, 0, "", 0)
120 wire = msg.to_wire()
121 msg2 = I2CPMessage.from_wire(wire)
122 assert msg2.reason == ""