distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for IRC client tunnel — TDD: tests before implementation."""
2
3import pytest
4
5from i2p_apps.i2ptunnel.config import TunnelDefinition, TunnelType
6from i2p_apps.i2ptunnel.irc_client import IRCClientTask
7
8
9class MockIRCSession:
10 def __init__(self):
11 self.destination = "IRC" * 172
12
13 async def lookup(self, name):
14 return "R" * 516
15
16 async def connect(self, dest):
17 import asyncio
18 r = asyncio.StreamReader()
19 r.feed_eof()
20 class FakeWriter:
21 def write(self, d): pass
22 async def drain(self): pass
23 def close(self): pass
24 async def wait_closed(self): pass
25 @property
26 def transport(self):
27 return type('T', (), {'is_closing': lambda: False})()
28 return r, FakeWriter()
29
30 async def close(self):
31 pass
32
33
34def _make_irc_def(**kw):
35 defaults = dict(
36 name="irc",
37 type=TunnelType.IRCCLIENT,
38 listen_port=6668,
39 target_destination="irc.postman.i2p",
40 )
41 defaults.update(kw)
42 return TunnelDefinition(**defaults)
43
44
45class TestIRCClientTask:
46 def test_creation(self):
47 task = IRCClientTask(_make_irc_def(), MockIRCSession())
48 assert task.tunnel_type == TunnelType.IRCCLIENT
49
50 def test_multi_dest_parsing(self):
51 task = IRCClientTask(
52 _make_irc_def(target_destination="irc1.i2p,irc2.i2p"),
53 MockIRCSession(),
54 )
55 assert len(task._destinations) == 2
56
57 def test_has_filtering(self):
58 """IRC client task should apply protocol filtering."""
59 task = IRCClientTask(_make_irc_def(), MockIRCSession())
60 assert task._uses_irc_filter is True