distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SessionTag and Payload data structures.
2
3TDD: these tests are written before the implementation.
4"""
5
6from __future__ import annotations
7
8import io
9import os
10import struct
11
12import pytest
13
14from i2p_data.session import SessionTag, Payload
15
16
17# ---------------------------------------------------------------------------
18# SessionTag
19# ---------------------------------------------------------------------------
20
21
22class TestSessionTag:
23 """Tests for the 32-byte SessionTag."""
24
25 def test_construction_valid(self):
26 data = os.urandom(32)
27 tag = SessionTag(data)
28 assert tag.to_bytes() == data
29
30 def test_construction_wrong_size_short(self):
31 with pytest.raises(ValueError, match="32 bytes"):
32 SessionTag(b"\x00" * 31)
33
34 def test_construction_wrong_size_long(self):
35 with pytest.raises(ValueError, match="32 bytes"):
36 SessionTag(b"\x00" * 33)
37
38 def test_construction_empty(self):
39 with pytest.raises(ValueError, match="32 bytes"):
40 SessionTag(b"")
41
42 def test_from_bytes(self):
43 data = os.urandom(32)
44 tag = SessionTag.from_bytes(data)
45 assert tag.to_bytes() == data
46
47 def test_from_bytes_wrong_size(self):
48 with pytest.raises(ValueError, match="32 bytes"):
49 SessionTag.from_bytes(b"\x00" * 10)
50
51 def test_random(self):
52 tag = SessionTag.random()
53 assert len(tag.to_bytes()) == 32
54
55 def test_random_unique(self):
56 """Two random tags should (almost certainly) differ."""
57 t1 = SessionTag.random()
58 t2 = SessionTag.random()
59 assert t1 != t2
60
61 def test_round_trip(self):
62 tag = SessionTag.random()
63 restored = SessionTag.from_bytes(tag.to_bytes())
64 assert restored == tag
65
66 def test_equality_same_data(self):
67 data = os.urandom(32)
68 assert SessionTag(data) == SessionTag(data)
69
70 def test_equality_different_data(self):
71 assert SessionTag(b"\x00" * 32) != SessionTag(b"\x01" * 32)
72
73 def test_equality_wrong_type(self):
74 tag = SessionTag(b"\x00" * 32)
75 assert tag != "not a tag"
76 assert tag != 42
77
78 def test_hash_same_data(self):
79 data = os.urandom(32)
80 assert hash(SessionTag(data)) == hash(SessionTag(data))
81
82 def test_hash_usable_in_set(self):
83 data = os.urandom(32)
84 s = {SessionTag(data), SessionTag(data)}
85 assert len(s) == 1
86
87 def test_repr(self):
88 tag = SessionTag(b"\xab\xcd" + b"\x00" * 30)
89 r = repr(tag)
90 assert "SessionTag" in r
91 assert "abcd" in r
92
93
94# ---------------------------------------------------------------------------
95# Payload
96# ---------------------------------------------------------------------------
97
98
99class TestPayload:
100 """Tests for the length-prefixed Payload."""
101
102 def test_construction(self):
103 p = Payload(b"hello")
104 assert len(p) == 5
105
106 def test_to_bytes_length_prefix(self):
107 p = Payload(b"hello")
108 wire = p.to_bytes()
109 assert wire[:4] == struct.pack("!I", 5)
110 assert wire[4:] == b"hello"
111
112 def test_from_bytes_round_trip(self):
113 original = Payload(b"test data")
114 wire = original.to_bytes()
115 restored = Payload.from_bytes(wire)
116 assert restored == original
117
118 def test_from_stream_round_trip(self):
119 original = Payload(b"stream test")
120 wire = original.to_bytes()
121 stream = io.BytesIO(wire)
122 restored = Payload.from_stream(stream)
123 assert restored == original
124
125 def test_empty_payload(self):
126 p = Payload(b"")
127 assert len(p) == 0
128 wire = p.to_bytes()
129 assert wire == struct.pack("!I", 0)
130 restored = Payload.from_bytes(wire)
131 assert restored == p
132
133 def test_large_payload(self):
134 data = os.urandom(100_000)
135 p = Payload(data)
136 assert len(p) == 100_000
137 wire = p.to_bytes()
138 restored = Payload.from_bytes(wire)
139 assert restored == p
140
141 def test_from_bytes_truncated_header(self):
142 with pytest.raises(ValueError, match="[Tt]runcated|[Nn]eed|bytes"):
143 Payload.from_bytes(b"\x00\x00")
144
145 def test_from_bytes_truncated_body(self):
146 # Header says 10 bytes, but only 3 are present
147 wire = struct.pack("!I", 10) + b"\x00\x00\x00"
148 with pytest.raises(ValueError, match="[Tt]runcated|[Nn]eed|bytes"):
149 Payload.from_bytes(wire)
150
151 def test_from_stream_truncated_header(self):
152 stream = io.BytesIO(b"\x00\x00")
153 with pytest.raises(ValueError, match="[Tt]runcated|[Nn]eed|bytes"):
154 Payload.from_stream(stream)
155
156 def test_from_stream_truncated_body(self):
157 wire = struct.pack("!I", 10) + b"\x00\x00\x00"
158 stream = io.BytesIO(wire)
159 with pytest.raises(ValueError, match="[Tt]runcated|[Nn]eed|bytes"):
160 Payload.from_stream(stream)
161
162 def test_equality_same(self):
163 assert Payload(b"abc") == Payload(b"abc")
164
165 def test_equality_different(self):
166 assert Payload(b"abc") != Payload(b"xyz")
167
168 def test_equality_wrong_type(self):
169 assert Payload(b"abc") != "abc"
170
171 def test_len(self):
172 assert len(Payload(b"1234")) == 4