distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for RouterWatchdog lifecycle and health checks."""
2
3import asyncio
4from unittest.mock import MagicMock
5
6import pytest
7
8from i2p_router.watchdog import RouterWatchdog
9
10
11def _make_router(state="running", uptime=100.0, jq=None, tm=None):
12 router = MagicMock()
13 router.state = state
14 router.uptime_seconds = uptime
15 ctx = MagicMock()
16 ctx.job_queue = jq
17 ctx.transport_manager = tm
18 router._context = ctx
19 return router
20
21
22class TestWatchdogProperties:
23 def test_initial_state(self):
24 wd = RouterWatchdog(_make_router())
25 assert wd.consecutive_errors == 0
26 assert not wd.is_running
27
28 def test_constants(self):
29 assert RouterWatchdog.CHECK_INTERVAL_SECONDS == 60
30 assert RouterWatchdog.MAX_CONSECUTIVE_ERRORS == 20
31
32
33class TestJobQueueLiveness:
34 def test_no_context(self):
35 router = MagicMock(spec=[])
36 wd = RouterWatchdog(router)
37 assert wd._verify_job_queue_liveness() is False
38
39 def test_no_job_queue(self):
40 router = _make_router(jq=None)
41 router._context.job_queue = None
42 wd = RouterWatchdog(router)
43 assert wd._verify_job_queue_liveness() is True
44
45 def test_never_run(self):
46 jq = MagicMock()
47 jq.last_job_run_time = 0
48 wd = RouterWatchdog(_make_router(jq=jq))
49 assert wd._verify_job_queue_liveness() is True
50
51 def test_recent_run(self):
52 import time
53 jq = MagicMock()
54 jq.last_job_run_time = time.monotonic() - 10 # 10s ago
55 wd = RouterWatchdog(_make_router(jq=jq))
56 assert wd._verify_job_queue_liveness() is True
57
58 def test_stale_run(self):
59 import time
60 jq = MagicMock()
61 jq.last_job_run_time = time.monotonic() - 1000 # way old
62 wd = RouterWatchdog(_make_router(jq=jq))
63 assert wd._verify_job_queue_liveness() is False
64
65
66class TestTransportLiveness:
67 def test_no_context(self):
68 router = MagicMock(spec=[])
69 wd = RouterWatchdog(router)
70 assert wd._verify_transport_liveness() is False
71
72 def test_no_transport_manager(self):
73 router = _make_router(tm=None)
74 router._context.transport_manager = None
75 wd = RouterWatchdog(router)
76 assert wd._verify_transport_liveness() is True
77
78 def test_transport_running(self):
79 tm = MagicMock()
80 tm.is_running = True
81 wd = RouterWatchdog(_make_router(tm=tm))
82 assert wd._verify_transport_liveness() is True
83
84 def test_transport_stopped(self):
85 tm = MagicMock()
86 tm.is_running = False
87 wd = RouterWatchdog(_make_router(tm=tm))
88 assert wd._verify_transport_liveness() is False
89
90
91class TestDumpStatus:
92 def test_includes_basic_info(self):
93 wd = RouterWatchdog(_make_router(state="running", uptime=500.0))
94 status = wd.dump_status()
95 assert "RouterWatchdog Status" in status
96 assert "running" in status
97 assert "500" in status
98
99 def test_includes_job_queue_info(self):
100 jq = MagicMock()
101 jq.pending_count = 42
102 wd = RouterWatchdog(_make_router(jq=jq))
103 status = wd.dump_status()
104 assert "42" in status
105
106 def test_includes_transport_info(self):
107 tm = MagicMock()
108 tm.is_running = True
109 wd = RouterWatchdog(_make_router(tm=tm))
110 status = wd.dump_status()
111 assert "True" in status
112
113
114class TestWatchdogLoop:
115 @pytest.mark.asyncio
116 async def test_run_and_stop(self):
117 wd = RouterWatchdog(_make_router())
118 wd.CHECK_INTERVAL_SECONDS = 0.05
119
120 task = asyncio.create_task(wd.run())
121 await asyncio.sleep(0.15)
122 wd.stop()
123 await asyncio.wait_for(task, timeout=2.0)
124 assert not wd.is_running
125
126 @pytest.mark.asyncio
127 async def test_consecutive_errors_reset_on_success(self):
128 import time
129 # Use a property that always returns current time
130 jq = MagicMock()
131 type(jq).last_job_run_time = property(lambda self: time.monotonic())
132 tm = MagicMock()
133 tm.is_running = True
134 wd = RouterWatchdog(_make_router(jq=jq, tm=tm))
135 wd.CHECK_INTERVAL_SECONDS = 0.05
136
137 task = asyncio.create_task(wd.run())
138 await asyncio.sleep(0.15)
139 wd.stop()
140 await asyncio.wait_for(task, timeout=2.0)
141 assert wd.consecutive_errors == 0
142
143 @pytest.mark.asyncio
144 async def test_errors_accumulate(self):
145 router = MagicMock(spec=[]) # no _context → both checks fail
146 wd = RouterWatchdog(router)
147 wd.CHECK_INTERVAL_SECONDS = 0.05
148
149 task = asyncio.create_task(wd.run())
150 await asyncio.sleep(0.2)
151 wd.stop()
152 await asyncio.wait_for(task, timeout=2.0)
153 assert wd.consecutive_errors > 0