distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for i2p_data.data_helper — I2P wire format serialization utilities."""
2import io
3from datetime import datetime, timezone
4
5import pytest
6
7from i2p_data.data_helper import (
8 from_base32,
9 from_base64,
10 from_date,
11 read_long,
12 read_properties,
13 read_string,
14 to_base32,
15 to_base64,
16 to_date,
17 write_long,
18 write_properties,
19 write_string,
20)
21
22
23# ---------------------------------------------------------------------------
24# read_long / write_long
25# ---------------------------------------------------------------------------
26
27class TestReadWriteLong:
28 """Roundtrip and edge-case tests for read_long / write_long."""
29
30 @pytest.mark.parametrize("num_bytes,value", [
31 (1, 0),
32 (1, 127),
33 (1, 255),
34 (2, 0),
35 (2, 256),
36 (2, 0xFFFF),
37 (4, 0),
38 (4, 1),
39 (4, 0xFFFFFFFF),
40 (8, 0),
41 (8, 1),
42 (8, 0xFFFFFFFFFFFFFFFF),
43 ])
44 def test_roundtrip(self, num_bytes: int, value: int) -> None:
45 buf = io.BytesIO()
46 write_long(buf, num_bytes, value)
47 buf.seek(0)
48 assert read_long(buf, num_bytes) == value
49
50 def test_read_known_bytes(self) -> None:
51 """Read a known big-endian 2-byte value: 0x0102 == 258."""
52 buf = io.BytesIO(b"\x01\x02")
53 assert read_long(buf, 2) == 258
54
55 def test_write_known_bytes(self) -> None:
56 """Write 258 as 2-byte big-endian and verify raw bytes."""
57 buf = io.BytesIO()
58 write_long(buf, 2, 258)
59 assert buf.getvalue() == b"\x01\x02"
60
61 def test_read_short_stream(self) -> None:
62 """Reading more bytes than available raises ValueError."""
63 buf = io.BytesIO(b"\x01")
64 with pytest.raises(ValueError, match="Expected 2 bytes, got 1"):
65 read_long(buf, 2)
66
67 def test_read_empty_stream(self) -> None:
68 buf = io.BytesIO(b"")
69 with pytest.raises(ValueError, match="Expected 1 bytes, got 0"):
70 read_long(buf, 1)
71
72 def test_sequential_writes_and_reads(self) -> None:
73 """Multiple values written sequentially are read back correctly."""
74 buf = io.BytesIO()
75 write_long(buf, 1, 0xAB)
76 write_long(buf, 4, 0xDEADBEEF)
77 write_long(buf, 2, 0x1234)
78 buf.seek(0)
79 assert read_long(buf, 1) == 0xAB
80 assert read_long(buf, 4) == 0xDEADBEEF
81 assert read_long(buf, 2) == 0x1234
82
83
84# ---------------------------------------------------------------------------
85# read_string / write_string
86# ---------------------------------------------------------------------------
87
88class TestReadWriteString:
89 """Roundtrip and edge-case tests for read_string / write_string."""
90
91 def test_empty_string(self) -> None:
92 buf = io.BytesIO()
93 write_string(buf, "")
94 buf.seek(0)
95 assert read_string(buf) == ""
96
97 def test_ascii_roundtrip(self) -> None:
98 buf = io.BytesIO()
99 write_string(buf, "hello")
100 buf.seek(0)
101 assert read_string(buf) == "hello"
102
103 def test_utf8_multibyte(self) -> None:
104 """Multi-byte UTF-8 characters (length prefix counts bytes, not chars)."""
105 text = "\u00e9\u00e8\u00ea" # e-acute, e-grave, e-circumflex (2 bytes each)
106 buf = io.BytesIO()
107 write_string(buf, text)
108 raw = buf.getvalue()
109 # Length prefix should be 6 (3 chars x 2 bytes each)
110 assert raw[0] == 6
111 buf.seek(0)
112 assert read_string(buf) == text
113
114 def test_max_length_string(self) -> None:
115 """String of exactly 255 bytes is accepted."""
116 text = "a" * 255
117 buf = io.BytesIO()
118 write_string(buf, text)
119 buf.seek(0)
120 assert read_string(buf) == text
121
122 def test_too_long_string(self) -> None:
123 """String longer than 255 bytes raises ValueError."""
124 text = "a" * 256
125 with pytest.raises(ValueError, match="String too long"):
126 buf = io.BytesIO()
127 write_string(buf, text)
128
129 def test_wire_format(self) -> None:
130 """Verify raw wire format: 1-byte length + UTF-8 payload."""
131 buf = io.BytesIO()
132 write_string(buf, "AB")
133 assert buf.getvalue() == b"\x02AB"
134
135 def test_read_truncated_string(self) -> None:
136 """Truncated payload raises ValueError."""
137 # Length says 5 but only 2 bytes of payload
138 buf = io.BytesIO(b"\x05AB")
139 with pytest.raises(ValueError, match="Expected 5 bytes for string, got 2"):
140 read_string(buf)
141
142
143# ---------------------------------------------------------------------------
144# read_properties / write_properties
145# ---------------------------------------------------------------------------
146
147class TestReadWriteProperties:
148 """Roundtrip and edge-case tests for read_properties / write_properties."""
149
150 def test_empty_dict(self) -> None:
151 buf = io.BytesIO()
152 write_properties(buf, {})
153 buf.seek(0)
154 assert read_properties(buf) == {}
155
156 def test_empty_dict_wire_format(self) -> None:
157 """Empty properties writes 2-byte zero length."""
158 buf = io.BytesIO()
159 write_properties(buf, {})
160 assert buf.getvalue() == b"\x00\x00"
161
162 def test_single_property(self) -> None:
163 buf = io.BytesIO()
164 write_properties(buf, {"key": "value"})
165 buf.seek(0)
166 assert read_properties(buf) == {"key": "value"}
167
168 def test_multiple_properties(self) -> None:
169 props = {"alpha": "1", "beta": "2", "gamma": "3"}
170 buf = io.BytesIO()
171 write_properties(buf, props)
172 buf.seek(0)
173 assert read_properties(buf) == props
174
175 def test_sorted_key_order(self) -> None:
176 """Keys are written in sorted order."""
177 props = {"z": "last", "a": "first", "m": "middle"}
178 buf = io.BytesIO()
179 write_properties(buf, props)
180 raw = buf.getvalue()
181 # Skip 2-byte length prefix, decode payload
182 payload = raw[2:].decode("utf-8")
183 lines = [l for l in payload.split("\n") if l.strip()]
184 keys = [l.split("=")[0] for l in lines]
185 assert keys == ["a", "m", "z"]
186
187 def test_value_with_equals(self) -> None:
188 """Values containing '=' are preserved (split on first '=' only)."""
189 props = {"url": "http://example.com?a=1&b=2"}
190 buf = io.BytesIO()
191 write_properties(buf, props)
192 buf.seek(0)
193 assert read_properties(buf) == props
194
195 def test_wire_format_semicolons(self) -> None:
196 """Each property line ends with ';\\n' in the wire format."""
197 buf = io.BytesIO()
198 write_properties(buf, {"k": "v"})
199 raw = buf.getvalue()
200 payload = raw[2:].decode("utf-8")
201 assert payload == "k=v;\n"
202
203 def test_read_truncated_properties(self) -> None:
204 """Truncated payload raises ValueError."""
205 # Length says 100 but we only have 3 bytes
206 buf = io.BytesIO(b"\x00\x64abc")
207 with pytest.raises(ValueError, match="Expected 100 bytes for properties"):
208 read_properties(buf)
209
210
211# ---------------------------------------------------------------------------
212# to_date / from_date
213# ---------------------------------------------------------------------------
214
215class TestDateConversion:
216 """Tests for to_date / from_date timestamp conversion."""
217
218 def test_epoch_zero(self) -> None:
219 dt = to_date(0)
220 assert dt == datetime(1970, 1, 1, tzinfo=timezone.utc)
221 assert from_date(dt) == 0
222
223 def test_known_timestamp(self) -> None:
224 """2025-01-01T00:00:00Z == 1735689600000 ms."""
225 ms = 1735689600000
226 dt = to_date(ms)
227 assert dt.year == 2025
228 assert dt.month == 1
229 assert dt.day == 1
230 assert from_date(dt) == ms
231
232 def test_roundtrip(self) -> None:
233 now = datetime(2026, 3, 17, 12, 0, 0, tzinfo=timezone.utc)
234 ms = from_date(now)
235 assert to_date(ms) == now
236
237 def test_sub_second_precision(self) -> None:
238 """Millisecond precision is preserved."""
239 ms = 1735689600123
240 dt = to_date(ms)
241 recovered = from_date(dt)
242 assert recovered == ms
243
244
245# ---------------------------------------------------------------------------
246# to_base64 / from_base64
247# ---------------------------------------------------------------------------
248
249class TestBase64:
250 """Tests for I2P-style Base64 encoding/decoding."""
251
252 def test_empty(self) -> None:
253 assert to_base64(b"") == ""
254 assert from_base64("") == b""
255
256 def test_roundtrip_short(self) -> None:
257 data = b"\x00\x01\x02\x03"
258 assert from_base64(to_base64(data)) == data
259
260 def test_roundtrip_long(self) -> None:
261 data = bytes(range(256))
262 assert from_base64(to_base64(data)) == data
263
264 def test_padding_in_output(self) -> None:
265 """I2P Base64 preserves '=' padding."""
266 # 1 byte encodes to 2 Base64 chars + padding
267 encoded = to_base64(b"\x00")
268 assert encoded == "AA=="
269
270 def test_known_vector(self) -> None:
271 """'Hello' -> 'SGVsbG8=' (I2P Base64 with padding)."""
272 assert to_base64(b"Hello") == "SGVsbG8="
273 assert from_base64("SGVsbG8") == b"Hello"
274 assert from_base64("SGVsbG8=") == b"Hello"
275
276 def test_binary_data(self) -> None:
277 data = b"\xff\xfe\xfd\xfc"
278 assert from_base64(to_base64(data)) == data
279
280
281# ---------------------------------------------------------------------------
282# to_base32 / from_base32
283# ---------------------------------------------------------------------------
284
285class TestBase32:
286 """Tests for Base32 encoding/decoding."""
287
288 def test_empty(self) -> None:
289 assert to_base32(b"") == ""
290 assert from_base32("") == b""
291
292 def test_roundtrip(self) -> None:
293 data = b"test data for base32"
294 assert from_base32(to_base32(data)) == data
295
296 def test_lowercase_output(self) -> None:
297 """I2P Base32 output is lowercase."""
298 encoded = to_base32(b"\xff\x00")
299 assert encoded == encoded.lower()
300
301 def test_no_padding(self) -> None:
302 encoded = to_base32(b"\x00\x01\x02")
303 assert "=" not in encoded
304
305 def test_known_vector(self) -> None:
306 """'Hi' -> base32 'jbuq' (lowercase, no padding)."""
307 encoded = to_base32(b"Hi")
308 assert encoded == "jbuq"
309 assert from_base32("jbuq") == b"Hi"
310
311 def test_accepts_mixed_case_input(self) -> None:
312 """from_base32 should accept both upper and lower case."""
313 data = b"CaSe"
314 encoded = to_base32(data)
315 assert from_base32(encoded.upper()) == data
316 assert from_base32(encoded.lower()) == data