distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for i2p_util core module — logging, hex, system, cache, collections, random, siphash."""
2
3import struct
4import threading
5
6import pytest
7
8
9# === Log / LogManager ===
10
11class TestLog:
12 def test_level_constants(self):
13 from i2p_util.log import Log
14 assert Log.DEBUG < Log.INFO < Log.WARN < Log.ERROR < Log.CRIT
15
16 def test_get_level(self):
17 from i2p_util.log import Log
18 assert Log.get_level("DEBUG") == Log.DEBUG
19 assert Log.get_level("info") == Log.INFO
20 assert Log.get_level("WARN") == Log.WARN
21 assert Log.get_level("ERROR") == Log.ERROR
22 assert Log.get_level("CRIT") == Log.CRIT
23 assert Log.get_level("CRITICAL") == Log.CRIT
24 # Unknown defaults to DEBUG
25 assert Log.get_level("bogus") == Log.DEBUG
26
27 def test_to_level_string(self):
28 from i2p_util.log import Log
29 assert Log.to_level_string(Log.DEBUG) == "DEBUG"
30 assert Log.to_level_string(Log.INFO) == "INFO"
31 assert Log.to_level_string(Log.WARN) == "WARN"
32 assert Log.to_level_string(Log.ERROR) == "ERROR"
33 assert Log.to_level_string(Log.CRIT) == "CRIT"
34 assert Log.to_level_string(999) == "DEBUG"
35
36 def test_should_log_respects_priority(self):
37 from i2p_util.log import Log, LogManager
38 mgr = LogManager()
39 log = mgr.get_log("test.priority")
40 log.set_minimum_priority(Log.WARN)
41 assert not log.should_log(Log.DEBUG)
42 assert not log.should_log(Log.INFO)
43 assert log.should_log(Log.WARN)
44 assert log.should_log(Log.ERROR)
45 assert log.should_log(Log.CRIT)
46
47 def test_should_debug_info_warn_error(self):
48 from i2p_util.log import Log, LogManager
49 mgr = LogManager()
50 log = mgr.get_log("test.shorthand")
51 log.set_minimum_priority(Log.ERROR)
52 assert not log.should_debug()
53 assert not log.should_info()
54 assert not log.should_warn()
55 assert log.should_error()
56
57 def test_log_does_not_raise(self):
58 from i2p_util.log import Log, LogManager
59 mgr = LogManager()
60 log = mgr.get_log("test.noexc")
61 log.set_minimum_priority(Log.DEBUG)
62 # Should not raise even with exception
63 log.debug("msg")
64 log.info("msg")
65 log.warn("msg")
66 log.error("msg")
67 log.log(Log.CRIT, "critical msg")
68 log.log(Log.ERROR, "with exc", ValueError("boom"))
69 log.log_always(Log.DEBUG, "forced")
70
71 def test_log_name(self):
72 from i2p_util.log import LogManager
73 mgr = LogManager()
74 log = mgr.get_log("my.logger")
75 assert log.name == "my.logger"
76
77 def test_log_from_class(self):
78 from i2p_util.log import LogManager
79 mgr = LogManager()
80 log = mgr.get_log(str)
81 assert "str" in log.name
82
83
84class TestLogManager:
85 def test_get_log_returns_same_instance(self):
86 from i2p_util.log import LogManager
87 mgr = LogManager()
88 a = mgr.get_log("same")
89 b = mgr.get_log("same")
90 assert a is b
91
92 def test_get_logs(self):
93 from i2p_util.log import LogManager
94 mgr = LogManager()
95 mgr.get_log("a")
96 mgr.get_log("b")
97 assert len(mgr.get_logs()) >= 2
98
99 def test_set_default_limit(self):
100 from i2p_util.log import Log, LogManager
101 mgr = LogManager()
102 log = mgr.get_log("default.test")
103 mgr.set_default_limit(Log.ERROR)
104 assert log.get_minimum_priority() == Log.ERROR
105 assert mgr.get_default_limit() == Log.ERROR
106
107 def test_shutdown_clears(self):
108 from i2p_util.log import LogManager
109 mgr = LogManager()
110 mgr.get_log("x")
111 mgr.shutdown()
112 assert len(mgr.get_logs()) == 0
113
114
115# === HexDump ===
116
117class TestHexDump:
118 def test_to_hex(self):
119 from i2p_util.hex import HexDump
120 assert HexDump.to_hex(b"\xde\xad\xbe\xef") == "deadbeef"
121
122 def test_from_hex(self):
123 from i2p_util.hex import HexDump
124 assert HexDump.from_hex("deadbeef") == b"\xde\xad\xbe\xef"
125
126 def test_roundtrip(self):
127 from i2p_util.hex import HexDump
128 data = b"Hello, I2P!"
129 assert HexDump.from_hex(HexDump.to_hex(data)) == data
130
131 def test_dump_format(self):
132 from i2p_util.hex import HexDump
133 data = b"ABCD"
134 out = HexDump.dump(data)
135 assert "00000000" in out
136 assert "41 42 43 44" in out
137 assert "|ABCD|" in out
138
139 def test_dump_non_printable(self):
140 from i2p_util.hex import HexDump
141 data = bytes(range(32))
142 out = HexDump.dump(data)
143 # Non-printable chars shown as dots
144 assert "." in out
145
146 def test_dump_offset_length(self):
147 from i2p_util.hex import HexDump
148 data = b"0123456789"
149 out = HexDump.dump(data, offset=2, length=4)
150 # Should show "2345"
151 assert "32 33 34 35" in out
152
153 def test_empty(self):
154 from i2p_util.hex import HexDump
155 assert HexDump.to_hex(b"") == ""
156 assert HexDump.from_hex("") == b""
157 assert HexDump.dump(b"") == ""
158
159
160# === SystemVersion ===
161
162class TestSystemVersion:
163 def test_get_os_returns_string(self):
164 from i2p_util.system import SystemVersion
165 os_name = SystemVersion.get_os()
166 assert isinstance(os_name, str)
167 assert len(os_name) > 0
168
169 def test_get_arch_returns_string(self):
170 from i2p_util.system import SystemVersion
171 arch = SystemVersion.get_arch()
172 assert arch in ("amd64", "arm64", "arm", "386", "unknown")
173
174 def test_is_64bit(self):
175 from i2p_util.system import SystemVersion
176 import struct
177 bits = struct.calcsize("P") * 8
178 assert SystemVersion.is_64bit() == (bits == 64)
179
180 def test_get_cores_positive(self):
181 from i2p_util.system import SystemVersion
182 assert SystemVersion.get_cores() >= 1
183
184 def test_get_max_memory_positive(self):
185 from i2p_util.system import SystemVersion
186 assert SystemVersion.get_max_memory() > 0
187
188 def test_is_android_false(self):
189 from i2p_util.system import SystemVersion
190 assert SystemVersion.is_android() is False
191
192 def test_python_version(self):
193 from i2p_util.system import SystemVersion
194 ver = SystemVersion.python_version()
195 assert "." in ver
196
197 def test_platform_booleans_consistent(self):
198 from i2p_util.system import SystemVersion
199 os_name = SystemVersion.get_os()
200 if os_name == "linux":
201 assert SystemVersion.is_linux()
202 elif os_name == "mac":
203 assert SystemVersion.is_mac()
204 elif os_name == "windows":
205 assert SystemVersion.is_windows()
206
207 def test_is_arm(self):
208 from i2p_util.system import SystemVersion
209 result = SystemVersion.is_arm()
210 assert isinstance(result, bool)
211
212 def test_is_x86(self):
213 from i2p_util.system import SystemVersion
214 result = SystemVersion.is_x86()
215 assert isinstance(result, bool)
216
217 def test_is_slow(self):
218 from i2p_util.system import SystemVersion
219 result = SystemVersion.is_slow()
220 assert isinstance(result, bool)
221
222 def test_get_os_returns_known(self):
223 from i2p_util.system import SystemVersion
224 from unittest.mock import patch
225 with patch("platform.system", return_value="Darwin"):
226 assert SystemVersion.get_os() == "mac"
227 with patch("platform.system", return_value="Windows"):
228 assert SystemVersion.get_os() == "windows"
229 with patch("platform.system", return_value="Linux"):
230 assert SystemVersion.get_os() == "linux"
231 with patch("platform.system", return_value="FreeBSD"):
232 assert SystemVersion.get_os() == "freebsd"
233
234 def test_get_arch_variants(self):
235 from i2p_util.system import SystemVersion
236 from unittest.mock import patch
237 with patch("platform.machine", return_value="x86_64"):
238 assert SystemVersion.get_arch() == "amd64"
239 with patch("platform.machine", return_value="aarch64"):
240 assert SystemVersion.get_arch() == "arm64"
241 with patch("platform.machine", return_value="armv7l"):
242 assert SystemVersion.get_arch() == "arm"
243 with patch("platform.machine", return_value="i686"):
244 assert SystemVersion.get_arch() == "386"
245 with patch("platform.machine", return_value="mips"):
246 assert SystemVersion.get_arch() == "unknown"
247
248 def test_is_windows(self):
249 from i2p_util.system import SystemVersion
250 from unittest.mock import patch
251 with patch("platform.system", return_value="Windows"):
252 assert SystemVersion.is_windows() is True
253 with patch("platform.system", return_value="Linux"):
254 assert SystemVersion.is_windows() is False
255
256 def test_is_mac(self):
257 from i2p_util.system import SystemVersion
258 from unittest.mock import patch
259 with patch("platform.system", return_value="Darwin"):
260 assert SystemVersion.is_mac() is True
261 with patch("platform.system", return_value="Linux"):
262 assert SystemVersion.is_mac() is False
263
264 def test_get_max_memory_fallback(self):
265 from i2p_util.system import SystemVersion
266 from unittest.mock import patch
267 with patch("resource.getrlimit", side_effect=ValueError("nope")):
268 result = SystemVersion.get_max_memory()
269 assert result == 2 * 1024 * 1024 * 1024
270
271
272# === LHMCache ===
273
274class TestLHMCache:
275 def test_basic_set_get(self):
276 from i2p_util.cache import LHMCache
277 c = LHMCache(3)
278 c["a"] = 1
279 assert c["a"] == 1
280
281 def test_eviction(self):
282 from i2p_util.cache import LHMCache
283 c = LHMCache(2)
284 c["a"] = 1
285 c["b"] = 2
286 c["c"] = 3 # Should evict "a"
287 assert "a" not in c
288 assert c["b"] == 2
289 assert c["c"] == 3
290
291 def test_access_refreshes(self):
292 from i2p_util.cache import LHMCache
293 c = LHMCache(2)
294 c["a"] = 1
295 c["b"] = 2
296 _ = c["a"] # Access "a" — moves it to end
297 c["c"] = 3 # Should evict "b", not "a"
298 assert "a" in c
299 assert "b" not in c
300
301 def test_update_refreshes(self):
302 from i2p_util.cache import LHMCache
303 c = LHMCache(2)
304 c["a"] = 1
305 c["b"] = 2
306 c["a"] = 10 # Update "a" — moves it to end
307 c["c"] = 3 # Should evict "b"
308 assert c["a"] == 10
309 assert "b" not in c
310
311 def test_get_default(self):
312 from i2p_util.cache import LHMCache
313 c = LHMCache(3)
314 assert c.get("missing") is None
315 assert c.get("missing", 42) == 42
316
317
318# === SimpleByteCache ===
319
320class TestSimpleByteCache:
321 def test_acquire_returns_bytearray(self):
322 from i2p_util.cache import SimpleByteCache
323 sbc = SimpleByteCache(4, 16)
324 buf = sbc.acquire()
325 assert isinstance(buf, bytearray)
326 assert len(buf) == 16
327
328 def test_release_and_reacquire(self):
329 from i2p_util.cache import SimpleByteCache
330 sbc = SimpleByteCache(4, 8)
331 buf = sbc.acquire()
332 buf[:] = b"\xff" * 8
333 sbc.release(buf)
334 # Re-acquired buffer should be zeroed
335 buf2 = sbc.acquire()
336 assert len(buf2) == 8
337
338 def test_pool_limit(self):
339 from i2p_util.cache import SimpleByteCache
340 sbc = SimpleByteCache(2, 4)
341 bufs = [sbc.acquire() for _ in range(5)]
342 for b in bufs:
343 sbc.release(b)
344 # Pool should only hold 2
345 assert len(sbc._pool) <= 2
346
347 def test_get_instance_singleton(self):
348 from i2p_util.cache import SimpleByteCache
349 # Clear class state first
350 a = SimpleByteCache.get_instance(64)
351 b = SimpleByteCache.get_instance(64)
352 assert a is b
353
354
355# === ByteCache ===
356
357class TestByteCache:
358 def test_acquire_release(self):
359 from i2p_util.cache import ByteCache
360 bc = ByteCache.get_instance(4, 32)
361 buf = bc.acquire()
362 assert len(buf) == 32
363 bc.release(buf)
364
365 def test_clear_all(self):
366 from i2p_util.cache import ByteCache
367 bc = ByteCache.get_instance(4, 32)
368 buf = bc.acquire()
369 bc.release(buf)
370 ByteCache.clear_all()
371
372
373# === OrderedProperties ===
374
375class TestOrderedProperties:
376 def test_sorted_keys(self):
377 from i2p_util.collections import OrderedProperties
378 p = OrderedProperties()
379 p["c"] = "3"
380 p["a"] = "1"
381 p["b"] = "2"
382 assert p.keys() == ["a", "b", "c"]
383
384 def test_sorted_items(self):
385 from i2p_util.collections import OrderedProperties
386 p = OrderedProperties()
387 p["z"] = "last"
388 p["a"] = "first"
389 items = p.items()
390 assert items[0] == ("a", "first")
391 assert items[1] == ("z", "last")
392
393 def test_get_set_property(self):
394 from i2p_util.collections import OrderedProperties
395 p = OrderedProperties()
396 p.set_property("key", "value")
397 assert p.get_property("key") == "value"
398 assert p.get_property("missing", "default") == "default"
399
400 def test_values_sorted_by_key(self):
401 from i2p_util.collections import OrderedProperties
402 p = OrderedProperties()
403 p["b"] = "B"
404 p["a"] = "A"
405 assert p.values() == ["A", "B"]
406
407
408# === RandomSource ===
409
410class TestRandomSource:
411 def test_singleton(self):
412 from i2p_util.random import RandomSource
413 a = RandomSource.get_instance()
414 b = RandomSource.get_instance()
415 assert a is b
416
417 def test_next_int_range(self):
418 from i2p_util.random import RandomSource
419 rs = RandomSource.get_instance()
420 for _ in range(100):
421 v = rs.next_int(10)
422 assert 0 <= v < 10
423
424 def test_next_int_bound_error(self):
425 from i2p_util.random import RandomSource
426 rs = RandomSource.get_instance()
427 with pytest.raises(ValueError):
428 rs.next_int(0)
429 with pytest.raises(ValueError):
430 rs.next_int(-1)
431
432 def test_signed_next_int_range(self):
433 from i2p_util.random import RandomSource
434 rs = RandomSource.get_instance()
435 v = rs.signed_next_int()
436 assert -(2**31) <= v < 2**31
437
438 def test_next_long(self):
439 from i2p_util.random import RandomSource
440 rs = RandomSource.get_instance()
441 for _ in range(50):
442 v = rs.next_long(1000)
443 assert 0 <= v < 1000
444
445 def test_next_bytes_length(self):
446 from i2p_util.random import RandomSource
447 rs = RandomSource.get_instance()
448 b = rs.next_bytes(32)
449 assert len(b) == 32
450
451 def test_next_bytes_into(self):
452 from i2p_util.random import RandomSource
453 rs = RandomSource.get_instance()
454 buf = bytearray(16)
455 rs.next_bytes_into(buf, 4, 8)
456 # First 4 and last 4 should still be zero
457 assert buf[:4] == b"\x00" * 4
458 assert buf[12:] == b"\x00" * 4
459 # Middle 8 bytes should be filled (overwhelmingly unlikely to be all zeros)
460
461 def test_next_boolean(self):
462 from i2p_util.random import RandomSource
463 rs = RandomSource.get_instance()
464 results = {rs.next_boolean() for _ in range(100)}
465 assert True in results
466 assert False in results
467
468
469# === SipHash ===
470
471class TestSipHash:
472 def test_digest_deterministic(self):
473 from i2p_util.siphash import SipHash
474 data = b"hello"
475 a = SipHash.digest(data)
476 b = SipHash.digest(data)
477 assert a == b
478
479 def test_digest_different_data(self):
480 from i2p_util.siphash import SipHash
481 a = SipHash.digest(b"hello")
482 b = SipHash.digest(b"world")
483 assert a != b
484
485 def test_digest_is_64bit(self):
486 from i2p_util.siphash import SipHash
487 h = SipHash.digest(b"test data")
488 assert 0 <= h < (1 << 64)
489
490 def test_hash_code_is_32bit(self):
491 from i2p_util.siphash import SipHash
492 h = SipHash.hash_code(b"test data")
493 assert 0 <= h < (1 << 32)
494
495 def test_hash_code_none(self):
496 from i2p_util.siphash import SipHash
497 assert SipHash.hash_code(None) == 0
498
499 def test_digest_with_offset_length(self):
500 from i2p_util.siphash import SipHash
501 data = b"XXXhelloXXX"
502 a = SipHash.digest(data, 3, 5)
503 b = SipHash.digest(b"hello")
504 assert a == b
505
506 def test_empty_data(self):
507 from i2p_util.siphash import SipHash
508 h = SipHash.digest(b"")
509 assert isinstance(h, int)
510
511 def test_various_lengths(self):
512 """Test all remainder cases (0-7 trailing bytes)."""
513 from i2p_util.siphash import SipHash
514 for length in range(17):
515 data = bytes(range(length))
516 h = SipHash.digest(data)
517 assert 0 <= h < (1 << 64)
518
519 def test_known_siphash24_vectors(self):
520 """Test against known SipHash-2-4 test vectors with k0=k1=0."""
521 from i2p_util.siphash import SipHash
522 # With known keys, we can verify the algorithm is correct
523 # Use internal method with k0=0, k1=0
524 result = SipHash._siphash24(
525 0x0706050403020100,
526 0x0f0e0d0c0b0a0908,
527 bytes(range(15)),
528 0, 15,
529 )
530 # Known test vector for SipHash-2-4 with these keys and this input
531 assert result == 0xa129ca6149be45e5