distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for i2p_time.clock — Clock offset management."""
2
3import time
4import threading
5from unittest.mock import MagicMock
6
7from i2p_time.clock import Clock, _system_millis
8from i2p_time.build_time import BuildTime
9
10
11class TestClock:
12 def test_now_returns_system_time_with_zero_offset(self):
13 clock = Clock()
14 before = _system_millis()
15 now = clock.now()
16 after = _system_millis()
17 assert before <= now <= after + 1 # allow 1ms tolerance
18
19 def test_get_offset_initially_zero_if_clock_ok(self):
20 clock = Clock()
21 # System clock should be within BuildTime bounds in normal operation
22 offset = clock.get_offset()
23 # If system clock is valid, offset is 0
24 if not clock._is_system_clock_bad:
25 assert offset == 0
26
27 def test_set_offset_changes_now(self):
28 clock = Clock()
29 clock.set_offset(5000, force=True)
30 now = clock.now()
31 sys_now = _system_millis()
32 # now() should be ~5000ms ahead of system time
33 assert abs((now - sys_now) - 5000) < 100
34
35 def test_set_offset_rejects_beyond_max(self):
36 clock = Clock()
37 clock._is_system_clock_bad = False
38 too_big = Clock.MAX_OFFSET + 1000
39 clock.set_offset(too_big)
40 assert clock.get_offset() == 0 # unchanged
41
42 def test_set_offset_force_bypasses_max(self):
43 clock = Clock()
44 clock._is_system_clock_bad = False
45 big = Clock.MAX_OFFSET + 1000
46 clock.set_offset(big, force=True)
47 assert clock.get_offset() == big
48
49 def test_set_offset_ignores_small_changes(self):
50 clock = Clock()
51 # First set to establish _already_changed
52 clock.set_offset(10000, force=True)
53 # Small change (< MIN_OFFSET_CHANGE) should be ignored
54 clock.set_offset(10000 + 100)
55 assert clock.get_offset() == 10000
56
57 def test_get_updated_successfully(self):
58 clock = Clock()
59 assert not clock.get_updated_successfully()
60 clock.set_offset(10000, force=True)
61 assert clock.get_updated_successfully()
62
63 def test_set_now_adjusts_offset(self):
64 clock = Clock()
65 # Set "real time" to 5 seconds ahead
66 real = _system_millis() + 5000
67 if BuildTime.get_earliest_time() <= real <= BuildTime.get_latest_time():
68 clock.set_now(real)
69 assert abs(clock.get_offset() - 5000) < 1000
70
71 def test_set_now_rejects_invalid_time(self):
72 clock = Clock()
73 clock.set_now(1000) # Way too early
74 assert clock.get_offset() == 0 # unchanged
75
76 def test_listener_called_on_offset_change(self):
77 clock = Clock()
78 listener = MagicMock()
79 clock.add_update_listener(listener)
80 clock.set_offset(10000, force=True)
81 listener.offset_changed.assert_called_once_with(10000)
82
83 def test_remove_listener(self):
84 clock = Clock()
85 listener = MagicMock()
86 clock.add_update_listener(listener)
87 clock.remove_update_listener(listener)
88 clock.set_offset(10000, force=True)
89 listener.offset_changed.assert_not_called()
90
91 def test_max_live_offset_after_startup(self):
92 clock = Clock()
93 # Simulate having been running for > 10 minutes
94 clock._already_changed = True
95 clock._started_on = _system_millis() - (11 * 60 * 1000)
96 clock._offset = 0
97 # Try to set a big offset (> MAX_LIVE_OFFSET)
98 clock.set_offset(Clock.MAX_LIVE_OFFSET + 1000)
99 assert clock.get_offset() == 0 # rejected
100
101
102class TestBuildTime:
103 def test_earliest_time_is_reasonable(self):
104 earliest = BuildTime.get_earliest_time()
105 # Should be sometime in 2025
106 assert earliest > 1700000000000 # > ~Nov 2023
107 assert earliest < 2000000000000 # < ~May 2033
108
109 def test_latest_is_after_earliest(self):
110 assert BuildTime.get_latest_time() > BuildTime.get_earliest_time()
111
112 def test_build_time_between_bounds(self):
113 assert BuildTime.get_earliest_time() <= BuildTime.get_build_time()
114 assert BuildTime.get_build_time() <= BuildTime.get_latest_time()