distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for SAMTunnelSession — TDD: tests before implementation.
2
3Uses MockSAMServer from conftest for protocol-level testing.
4"""
5
6import asyncio
7
8import pytest
9
10from i2p_apps.i2ptunnel.sam_session import SAMTunnelSession, SessionFactory
11
12
13@pytest.mark.asyncio
14class TestSAMTunnelSession:
15 async def test_create_session(self, mock_sam_server):
16 """Session creation via HELLO + SESSION CREATE."""
17 session = SAMTunnelSession(mock_sam_server.host, mock_sam_server.port)
18 await session.open(nickname="test-session")
19 assert session.destination is not None
20 assert len(session.destination) > 0
21 await session.close()
22
23 async def test_naming_lookup_found(self, mock_sam_server):
24 mock_sam_server.add_name("forum.i2p", "B" * 516)
25 session = SAMTunnelSession(mock_sam_server.host, mock_sam_server.port)
26 await session.open(nickname="lookup-test")
27 dest = await session.lookup("forum.i2p")
28 assert dest == "B" * 516
29 await session.close()
30
31 async def test_naming_lookup_not_found(self, mock_sam_server):
32 session = SAMTunnelSession(mock_sam_server.host, mock_sam_server.port)
33 await session.open(nickname="lookup-miss")
34 dest = await session.lookup("nonexistent.i2p")
35 assert dest is None
36 await session.close()
37
38 async def test_connect_returns_streams(self, mock_sam_server):
39 """STREAM CONNECT returns reader/writer pair."""
40 session = SAMTunnelSession(mock_sam_server.host, mock_sam_server.port)
41 await session.open(nickname="connect-test")
42 reader, writer = await session.connect("C" * 516)
43 assert reader is not None
44 assert writer is not None
45 writer.close()
46 await writer.wait_closed()
47 await session.close()
48
49 async def test_connect_echo(self, mock_sam_server):
50 """MockSAMServer echoes data after STREAM CONNECT."""
51 session = SAMTunnelSession(mock_sam_server.host, mock_sam_server.port)
52 await session.open(nickname="echo-test")
53 reader, writer = await session.connect("D" * 516)
54 writer.write(b"ping")
55 await writer.drain()
56 # Read echo before closing — server echoes data back
57 data = await asyncio.wait_for(reader.read(4096), timeout=2.0)
58 assert data == b"ping"
59 writer.close()
60 await session.close()
61
62 async def test_session_with_private_key(self, mock_sam_server):
63 """Session creation with a private key."""
64 session = SAMTunnelSession(mock_sam_server.host, mock_sam_server.port)
65 await session.open(nickname="key-test", private_key="E" * 516)
66 assert session.destination is not None
67 await session.close()
68
69 async def test_close_idempotent(self, mock_sam_server):
70 session = SAMTunnelSession(mock_sam_server.host, mock_sam_server.port)
71 await session.open(nickname="close-test")
72 await session.close()
73 await session.close() # should not raise
74
75
76@pytest.mark.asyncio
77class TestSessionFactory:
78 async def test_create_session(self, mock_sam_server):
79 factory = SessionFactory(mock_sam_server.host, mock_sam_server.port)
80 session = await factory.create(nickname="factory-test")
81 assert session.destination is not None
82 await factory.release(session)
83
84 async def test_shared_session(self, mock_sam_server):
85 """Shared sessions reuse the same SAM connection."""
86 factory = SessionFactory(mock_sam_server.host, mock_sam_server.port)
87 s1 = await factory.get_shared("shared-group")
88 s2 = await factory.get_shared("shared-group")
89 assert s1 is s2 # same object
90 await factory.release_shared("shared-group")