distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for HTTPProxy server — handle_client and error responses."""
2
3import asyncio
4from unittest.mock import AsyncMock, MagicMock
5
6import pytest
7
8from i2p_apps.i2ptunnel.http_proxy import HTTPProxy
9
10
11def _mock_writer():
12 writer = MagicMock()
13 writer.write = MagicMock()
14 writer.drain = AsyncMock()
15 writer.close = MagicMock()
16 return writer
17
18
19class TestHTTPProxyProperties:
20 def test_defaults(self):
21 proxy = HTTPProxy()
22 assert proxy.listen_address == ("127.0.0.1", 4444)
23 assert proxy.is_running is False
24
25 def test_custom(self):
26 proxy = HTTPProxy("0.0.0.0", 8080)
27 assert proxy.listen_address == ("0.0.0.0", 8080)
28
29 @pytest.mark.asyncio
30 async def test_start_stop(self):
31 proxy = HTTPProxy("127.0.0.1", 0)
32 await proxy.start()
33 assert proxy.is_running is True
34 await proxy.stop()
35 assert proxy.is_running is False
36
37 @pytest.mark.asyncio
38 async def test_stop_without_start(self):
39 proxy = HTTPProxy()
40 await proxy.stop()
41 assert proxy.is_running is False
42
43
44class TestHTTPProxyErrorResponse:
45 def test_build_error_response(self):
46 proxy = HTTPProxy()
47 resp = proxy.build_error_response(404, "Not Found")
48 text = resp.decode()
49 assert "HTTP/1.1 404 Not Found" in text
50 assert "Content-Type: text/html" in text
51 assert "<h1>404 Not Found</h1>" in text
52
53 def test_build_error_502(self):
54 proxy = HTTPProxy()
55 resp = proxy.build_error_response(502, "Bad Gateway")
56 assert b"502" in resp
57
58
59class TestHTTPProxyHandleClient:
60 @pytest.mark.asyncio
61 async def test_empty_read(self):
62 proxy = HTTPProxy()
63 reader = AsyncMock()
64 reader.readline = AsyncMock(return_value=b"")
65 writer = _mock_writer()
66 await proxy.handle_client(reader, writer)
67 writer.close.assert_called()
68
69 @pytest.mark.asyncio
70 async def test_i2p_request(self):
71 proxy = HTTPProxy()
72 reader = AsyncMock()
73 reader.readline = AsyncMock(return_value=b"GET http://example.i2p/ HTTP/1.1\r\n")
74 writer = _mock_writer()
75 await proxy.handle_client(reader, writer)
76 # Should return 502 (no I2P session)
77 written = writer.write.call_args[0][0]
78 assert b"502" in written
79
80 @pytest.mark.asyncio
81 async def test_non_i2p_request(self):
82 proxy = HTTPProxy()
83 reader = AsyncMock()
84 reader.readline = AsyncMock(return_value=b"GET http://example.com/ HTTP/1.1\r\n")
85 writer = _mock_writer()
86 await proxy.handle_client(reader, writer)
87 written = writer.write.call_args[0][0]
88 assert b"502" in written
89
90 @pytest.mark.asyncio
91 async def test_exception_closes_writer(self):
92 proxy = HTTPProxy()
93 reader = AsyncMock()
94 reader.readline = AsyncMock(side_effect=ConnectionResetError("reset"))
95 writer = _mock_writer()
96 await proxy.handle_client(reader, writer)
97 writer.close.assert_called()
98
99 @pytest.mark.asyncio
100 async def test_connect_request(self):
101 proxy = HTTPProxy()
102 reader = AsyncMock()
103 reader.readline = AsyncMock(return_value=b"CONNECT example.i2p:443 HTTP/1.1\r\n")
104 writer = _mock_writer()
105 await proxy.handle_client(reader, writer)
106 written = writer.write.call_args[0][0]
107 assert b"502" in written