distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for GenericServerTask — TDD: tests before implementation."""
2
3import pytest
4
5from i2p_apps.i2ptunnel.config import TunnelDefinition, TunnelType
6from i2p_apps.i2ptunnel.server_task import GenericServerTask
7
8
9def _make_server_def(**kw):
10 defaults = dict(
11 name="test-server",
12 type=TunnelType.SERVER,
13 target_host="127.0.0.1",
14 target_port=8080,
15 )
16 defaults.update(kw)
17 return TunnelDefinition(**defaults)
18
19
20class MockServerSession:
21 def __init__(self):
22 self.destination = "SERV" * 129
23
24 async def accept(self):
25 raise NotImplementedError("No mock accept")
26
27 async def close(self):
28 pass
29
30
31class TestGenericServerTask:
32 def test_creation(self):
33 config = _make_server_def()
34 session = MockServerSession()
35 task = GenericServerTask(config, session)
36 assert task.tunnel_type == TunnelType.SERVER
37 assert not task.is_open
38
39 def test_config_target(self):
40 config = _make_server_def(target_host="10.0.0.1", target_port=9090)
41 session = MockServerSession()
42 task = GenericServerTask(config, session)
43 assert task._config.target_host == "10.0.0.1"
44 assert task._config.target_port == 9090
45
46 def test_stats_initial(self):
47 config = _make_server_def()
48 session = MockServerSession()
49 task = GenericServerTask(config, session)
50 assert task.stats.total_connections == 0
51 assert task.stats.active_connections == 0
52 assert task.stats.total_bytes == 0