distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for bidirectional forwarder — TDD: tests before implementation."""
2
3import asyncio
4
5import pytest
6
7from i2p_apps.i2ptunnel.forwarder import forward, bridge
8
9
10def _make_streams(data: bytes = b""):
11 """Create an in-memory StreamReader pre-loaded with data, and a StreamWriter backed by a transport."""
12 reader = asyncio.StreamReader()
13 if data:
14 reader.feed_data(data)
15 reader.feed_eof()
16 return reader
17
18
19class _CapturingTransport:
20 """Minimal transport that captures written bytes."""
21
22 def __init__(self):
23 self.data = bytearray()
24 self._closing = False
25
26 def write(self, data):
27 self.data.extend(data)
28
29 def is_closing(self):
30 return self._closing
31
32 def close(self):
33 self._closing = True
34
35 def get_extra_info(self, key, default=None):
36 return default
37
38
39def _make_writer_pair():
40 """Return (writer, captured_bytes_transport)."""
41 transport = _CapturingTransport()
42 protocol = asyncio.StreamReaderProtocol(asyncio.StreamReader())
43 writer = asyncio.StreamWriter(transport, protocol, asyncio.StreamReader(), asyncio.get_event_loop())
44 return writer, transport
45
46
47class TestForward:
48 @pytest.mark.asyncio
49 async def test_forward_copies_data(self):
50 reader = _make_streams(b"hello world")
51 writer, transport = _make_writer_pair()
52 await forward(reader, writer)
53 assert bytes(transport.data) == b"hello world"
54
55 @pytest.mark.asyncio
56 async def test_forward_handles_empty(self):
57 reader = _make_streams(b"")
58 writer, transport = _make_writer_pair()
59 await forward(reader, writer)
60 assert bytes(transport.data) == b""
61
62 @pytest.mark.asyncio
63 async def test_forward_large_data(self):
64 data = b"x" * 50000
65 reader = _make_streams(data)
66 writer, transport = _make_writer_pair()
67 await forward(reader, writer)
68 assert bytes(transport.data) == data
69
70
71class TestBridge:
72 @pytest.mark.asyncio
73 async def test_bridge_bidirectional(self):
74 """Data flows in both directions."""
75 # A->B: reader_a has data, writer_b captures
76 reader_a = _make_streams(b"from-a")
77 writer_b, transport_b = _make_writer_pair()
78
79 # B->A: reader_b has data, writer_a captures
80 reader_b = _make_streams(b"from-b")
81 writer_a, transport_a = _make_writer_pair()
82
83 await bridge(reader_a, writer_a, reader_b, writer_b)
84 assert bytes(transport_b.data) == b"from-a"
85 assert bytes(transport_a.data) == b"from-b"
86
87 @pytest.mark.asyncio
88 async def test_bridge_one_side_empty(self):
89 reader_a = _make_streams(b"payload")
90 writer_b, transport_b = _make_writer_pair()
91
92 reader_b = _make_streams(b"")
93 writer_a, transport_a = _make_writer_pair()
94
95 await bridge(reader_a, writer_a, reader_b, writer_b)
96 assert bytes(transport_b.data) == b"payload"
97 assert bytes(transport_a.data) == b""