distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for MessageValidator."""
2
3import time
4from i2p_router.message_validator import MessageValidator
5
6
7def _now_ms():
8 return int(time.time() * 1000)
9
10
11def test_accept_valid_message():
12 mv = MessageValidator()
13 now = _now_ms()
14 assert mv.validate_message(1234, now + 30_000, now) is True
15
16
17def test_reject_expired():
18 mv = MessageValidator()
19 now = _now_ms()
20 # Expired 2 minutes ago (well beyond 1.5x fudge)
21 assert mv.validate_message(1, now - 120_000, now) is False
22
23
24def test_reject_far_future():
25 mv = MessageValidator()
26 now = _now_ms()
27 # 5 minutes in future (beyond 4x fudge = 260s)
28 assert mv.validate_message(1, now + 300_000, now) is False
29
30
31def test_reject_duplicate():
32 mv = MessageValidator()
33 now = _now_ms()
34 exp = now + 30_000
35 assert mv.validate_message(42, exp, now) is True
36 assert mv.validate_message(42, exp, now) is False # duplicate
37
38
39def test_near_expiration_accepted():
40 mv = MessageValidator()
41 now = _now_ms()
42 # Just barely within fudge
43 exp = now - 50_000 # expired 50s ago, fudge is 97.5s
44 assert mv.validate_message(99, exp, now) is True
45
46
47def test_different_id_same_expiration():
48 mv = MessageValidator()
49 now = _now_ms()
50 exp = now + 30_000
51 assert mv.validate_message(100, exp, now) is True
52 assert mv.validate_message(200, exp, now) is True # different ID