distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for TunnelController lifecycle — TDD: tests before implementation."""
2
3import asyncio
4
5import pytest
6
7from i2p_apps.i2ptunnel.config import TunnelDefinition, TunnelType
8from i2p_apps.i2ptunnel.controller import TunnelController, TunnelState
9
10
11def _make_def(**kw) -> TunnelDefinition:
12 defaults = dict(name="test", type=TunnelType.CLIENT, listen_port=0)
13 defaults.update(kw)
14 return TunnelDefinition(**defaults)
15
16
17class MockSession:
18 """Fake TunnelSession for controller tests."""
19
20 def __init__(self):
21 self.closed = False
22 self.destination = "AAAA" * 129 # fake base64
23
24 async def close(self):
25 self.closed = True
26
27 async def connect(self, dest):
28 return None, None
29
30 async def accept(self):
31 return "", None, None
32
33 async def lookup(self, name):
34 return None
35
36
37class MockSessionFactory:
38 """Factory that returns MockSession instances."""
39
40 def __init__(self):
41 self.sessions: list[MockSession] = []
42
43 async def create(self, tunnel_def, priv_key=None):
44 s = MockSession()
45 self.sessions.append(s)
46 return s
47
48 async def release(self, session):
49 session.closed = True
50
51
52class TestTunnelState:
53 def test_enum_values(self):
54 assert TunnelState.STOPPED.value == "stopped"
55 assert TunnelState.STARTING.value == "starting"
56 assert TunnelState.RUNNING.value == "running"
57 assert TunnelState.STOPPING.value == "stopping"
58 assert TunnelState.DESTROYED.value == "destroyed"
59
60
61class TestTunnelControllerLifecycle:
62 def test_initial_state_is_stopped(self):
63 tc = TunnelController(_make_def(), MockSessionFactory())
64 assert tc.state == TunnelState.STOPPED
65
66 def test_start_transitions_to_running(self):
67 tc = TunnelController(_make_def(), MockSessionFactory())
68 asyncio.run(tc.start())
69 assert tc.state == TunnelState.RUNNING
70
71 def test_stop_transitions_to_stopped(self):
72 tc = TunnelController(_make_def(), MockSessionFactory())
73 asyncio.run(tc.start())
74 asyncio.run(tc.stop())
75 assert tc.state == TunnelState.STOPPED
76
77 def test_restart(self):
78 tc = TunnelController(_make_def(), MockSessionFactory())
79 asyncio.run(tc.start())
80 asyncio.run(tc.restart())
81 assert tc.state == TunnelState.RUNNING
82
83 def test_destroy_prevents_restart(self):
84 tc = TunnelController(_make_def(), MockSessionFactory())
85 tc.destroy()
86 assert tc.state == TunnelState.DESTROYED
87 with pytest.raises(RuntimeError, match="destroyed"):
88 asyncio.run(tc.start())
89
90 def test_stop_when_stopped_is_noop(self):
91 tc = TunnelController(_make_def(), MockSessionFactory())
92 asyncio.run(tc.stop()) # should not raise
93 assert tc.state == TunnelState.STOPPED
94
95 def test_start_when_running_is_noop(self):
96 tc = TunnelController(_make_def(), MockSessionFactory())
97 asyncio.run(tc.start())
98 asyncio.run(tc.start()) # second start is noop
99 assert tc.state == TunnelState.RUNNING
100
101
102class TestTunnelControllerSession:
103 def test_session_created_on_start(self):
104 factory = MockSessionFactory()
105 tc = TunnelController(_make_def(), factory)
106 asyncio.run(tc.start())
107 assert len(factory.sessions) == 1
108
109 def test_session_released_on_stop(self):
110 factory = MockSessionFactory()
111 tc = TunnelController(_make_def(), factory)
112 asyncio.run(tc.start())
113 asyncio.run(tc.stop())
114 assert factory.sessions[0].closed is True
115
116 def test_config_accessible(self):
117 td = _make_def(name="mytest")
118 tc = TunnelController(td, MockSessionFactory())
119 assert tc.config.name == "mytest"