distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for Router lifecycle — state transitions, start, shutdown, status."""
2
3import time
4import pytest
5
6from i2p_router.router import Router, RouterState
7from i2p_router.config import RouterConfig
8
9
10class TestInitialState:
11 """Router initial state is STOPPED."""
12
13 def test_initial_state_is_stopped(self):
14 router = Router()
15 assert router.state == RouterState.STOPPED
16
17 def test_initial_uptime_is_zero(self):
18 router = Router()
19 assert router.uptime_seconds == 0.0
20
21 def test_initial_config_is_default(self):
22 router = Router()
23 assert router.config.router_name == "i2p-python-router"
24 assert router.config.listen_port == 9700
25
26 def test_custom_config(self):
27 cfg = RouterConfig(router_name="test-router", listen_port=8080)
28 router = Router(config=cfg)
29 assert router.config.router_name == "test-router"
30 assert router.config.listen_port == 8080
31
32
33class TestStartTransition:
34 """start() transitions to STARTING then RUNNING."""
35
36 def test_start_reaches_running(self):
37 router = Router()
38 router.start()
39 assert router.state == RouterState.RUNNING
40
41 def test_start_sets_uptime(self):
42 router = Router()
43 router.start()
44 # Uptime should be very small but non-negative
45 assert router.uptime_seconds >= 0.0
46
47 def test_start_creates_context(self):
48 router = Router()
49 router.start()
50 assert router._context is not None
51
52 def test_start_with_invalid_config_raises(self):
53 cfg = RouterConfig(listen_port=-1)
54 router = Router(config=cfg)
55 with pytest.raises(ValueError):
56 router.start()
57 # State should remain STOPPED on validation failure
58 assert router.state == RouterState.STOPPED
59
60
61class TestShutdownTransition:
62 """shutdown() transitions to SHUTTING_DOWN then STOPPED."""
63
64 def test_shutdown_reaches_stopped(self):
65 router = Router()
66 router.start()
67 assert router.state == RouterState.RUNNING
68 router.shutdown()
69 assert router.state == RouterState.STOPPED
70
71 def test_shutdown_clears_uptime(self):
72 router = Router()
73 router.start()
74 router.shutdown()
75 assert router.uptime_seconds == 0.0
76
77 def test_shutdown_when_already_stopped_is_noop(self):
78 router = Router()
79 router.shutdown() # Should not raise
80 assert router.state == RouterState.STOPPED
81
82
83class TestGetStatus:
84 """get_status() returns correct state info."""
85
86 def test_status_when_stopped(self):
87 router = Router()
88 status = router.get_status()
89 assert status["state"] == "stopped"
90 assert status["uptime_seconds"] == 0.0
91 assert "config" in status
92
93 def test_status_when_running(self):
94 router = Router()
95 router.start()
96 status = router.get_status()
97 assert status["state"] == "running"
98 assert status["uptime_seconds"] >= 0.0
99 assert status["config"]["router_name"] == "i2p-python-router"
100 router.shutdown()
101
102 def test_status_config_reflects_custom_config(self):
103 cfg = RouterConfig(router_name="custom", listen_port=7000)
104 router = Router(config=cfg)
105 status = router.get_status()
106 assert status["config"]["router_name"] == "custom"
107 assert status["config"]["listen_port"] == 7000
108
109
110class TestIdempotentStart:
111 """Double start() is idempotent."""
112
113 def test_double_start_stays_running(self):
114 router = Router()
115 router.start()
116 router.start() # Should not raise
117 assert router.state == RouterState.RUNNING
118 router.shutdown()
119
120 def test_double_start_preserves_context(self):
121 router = Router()
122 router.start()
123 ctx1 = router._context
124 router.start()
125 ctx2 = router._context
126 # Context should be the same object (not recreated)
127 assert ctx1 is ctx2
128 router.shutdown()
129
130
131class TestRestart:
132 """start after shutdown works (restart)."""
133
134 def test_restart_reaches_running(self):
135 router = Router()
136 router.start()
137 router.shutdown()
138 assert router.state == RouterState.STOPPED
139 router.start()
140 assert router.state == RouterState.RUNNING
141 router.shutdown()
142
143 def test_restart_resets_uptime(self):
144 router = Router()
145 router.start()
146 time.sleep(0.01)
147 uptime1 = router.uptime_seconds
148 router.shutdown()
149 router.start()
150 uptime2 = router.uptime_seconds
151 # New uptime should be less than old (fresh start)
152 assert uptime2 < uptime1
153 router.shutdown()
154
155 def test_restart_creates_new_context(self):
156 router = Router()
157 router.start()
158 ctx1 = router._context
159 router.shutdown()
160 router.start()
161 ctx2 = router._context
162 # After restart, new context is created
163 assert ctx1 is not ctx2
164 router.shutdown()