distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for net.i2p.util Tier 3 — HTTP client & SSL.
2
3TDD — tests for I2PSSLSocketFactory, EepGet, SSLEepGet.
4Uses a local test HTTP server, no external network access.
5"""
6
7from __future__ import annotations
8
9import http.server
10import json
11import ssl
12import tempfile
13import threading
14import os
15
16import pytest
17
18from i2p_util.ssl_factory import create_ssl_context
19from i2p_util.eep_get import EepGet, EepGetResult
20
21
22# ---------------------------------------------------------------------------
23# SSL Context Factory
24# ---------------------------------------------------------------------------
25
26
27class TestSSLFactory:
28 """SSL context creation."""
29
30 def test_create_default_context(self):
31 ctx = create_ssl_context()
32 assert isinstance(ctx, ssl.SSLContext)
33
34 def test_minimum_tls_version(self):
35 ctx = create_ssl_context()
36 assert ctx.minimum_version >= ssl.TLSVersion.TLSv1_2
37
38 def test_hostname_check_disabled_for_i2p(self):
39 ctx = create_ssl_context(verify_hostname=False)
40 assert ctx.check_hostname is False
41
42
43# ---------------------------------------------------------------------------
44# Test HTTP server fixture
45# ---------------------------------------------------------------------------
46
47
48class _TestHandler(http.server.BaseHTTPRequestHandler):
49 """Minimal test HTTP server."""
50
51 def do_GET(self):
52 if self.path == "/hello":
53 body = b"Hello, World!"
54 self.send_response(200)
55 self.send_header("Content-Length", str(len(body)))
56 self.end_headers()
57 self.wfile.write(body)
58 elif self.path == "/redirect":
59 self.send_response(302)
60 self.send_header("Location", "/hello")
61 self.end_headers()
62 elif self.path == "/large":
63 body = b"X" * 10000
64 self.send_response(200)
65 self.send_header("Content-Length", str(len(body)))
66 self.end_headers()
67 self.wfile.write(body)
68 elif self.path == "/not-modified":
69 ims = self.headers.get("If-Modified-Since")
70 if ims:
71 self.send_response(304)
72 self.end_headers()
73 else:
74 body = b"fresh content"
75 self.send_response(200)
76 self.send_header("Content-Length", str(len(body)))
77 self.end_headers()
78 self.wfile.write(body)
79 else:
80 self.send_response(404)
81 self.end_headers()
82
83 def log_message(self, format, *args):
84 pass # suppress logs
85
86
87@pytest.fixture(scope="module")
88def test_server():
89 """Start a local HTTP server for testing."""
90 server = http.server.HTTPServer(("127.0.0.1", 0), _TestHandler)
91 port = server.server_address[1]
92 thread = threading.Thread(target=server.serve_forever, daemon=True)
93 thread.start()
94 yield f"http://127.0.0.1:{port}"
95 server.shutdown()
96
97
98# ---------------------------------------------------------------------------
99# EepGet
100# ---------------------------------------------------------------------------
101
102
103class TestEepGet:
104 """HTTP client for I2P."""
105
106 def test_fetch_success(self, test_server):
107 eg = EepGet()
108 result = eg.fetch(f"{test_server}/hello")
109 assert result.success is True
110 assert result.data == b"Hello, World!"
111 assert result.status_code == 200
112
113 def test_fetch_404(self, test_server):
114 eg = EepGet()
115 result = eg.fetch(f"{test_server}/missing")
116 assert result.status_code == 404
117
118 def test_fetch_follow_redirect(self, test_server):
119 eg = EepGet()
120 result = eg.fetch(f"{test_server}/redirect")
121 assert result.success is True
122 assert result.data == b"Hello, World!"
123
124 def test_fetch_to_file(self, test_server):
125 eg = EepGet()
126 with tempfile.NamedTemporaryFile(delete=False) as f:
127 tmp_path = f.name
128 try:
129 result = eg.fetch(f"{test_server}/hello", output_file=tmp_path)
130 assert result.success is True
131 with open(tmp_path, "rb") as f:
132 assert f.read() == b"Hello, World!"
133 finally:
134 os.unlink(tmp_path)
135
136 def test_fetch_large(self, test_server):
137 eg = EepGet()
138 result = eg.fetch(f"{test_server}/large")
139 assert result.success is True
140 assert len(result.data) == 10000
141
142 def test_progress_callback(self, test_server):
143 progress_calls = []
144
145 def on_progress(downloaded, total):
146 progress_calls.append((downloaded, total))
147
148 eg = EepGet()
149 result = eg.fetch(f"{test_server}/large", progress_callback=on_progress)
150 assert result.success is True
151 assert len(progress_calls) > 0
152
153 def test_timeout(self):
154 """Connection to non-routable address should timeout."""
155 eg = EepGet(connect_timeout=0.5)
156 result = eg.fetch("http://192.0.2.1:1/timeout")
157 assert result.success is False
158
159 def test_if_modified_since(self, test_server):
160 eg = EepGet()
161 result = eg.fetch(
162 f"{test_server}/not-modified",
163 if_modified_since="Thu, 01 Jan 2099 00:00:00 GMT",
164 )
165 assert result.status_code == 304
166
167 def test_user_agent(self, test_server):
168 """Default User-Agent should be 'I2P'."""
169 eg = EepGet()
170 assert eg.user_agent == "I2P"
171
172
173# ---------------------------------------------------------------------------
174# EepGetResult
175# ---------------------------------------------------------------------------
176
177
178class TestEepGetResult:
179 """Result object from EepGet."""
180
181 def test_success_result(self):
182 r = EepGetResult(success=True, status_code=200, data=b"ok")
183 assert r.success
184 assert r.content_length == 2
185
186 def test_failure_result(self):
187 r = EepGetResult(success=False, status_code=0, data=b"", error="timeout")
188 assert not r.success
189 assert r.error == "timeout"