distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for network probes — TDD: tests before implementation."""
2
3import socket
4from unittest.mock import MagicMock, patch
5
6import pytest
7
8from i2p_apps.setup.network_probe import (
9 BandwidthEstimate,
10 DnsProbeResult,
11 LatencyProbeResult,
12 NetworkProbeResult,
13 check_dns,
14 estimate_bandwidth,
15 measure_tcp_rtt,
16 probe_dns,
17 probe_latency,
18 run_network_probes,
19)
20
21
22class TestCheckDns:
23 def test_successful_resolution(self):
24 with patch("i2p_apps.setup.network_probe.socket.getaddrinfo") as mock:
25 mock.return_value = [(2, 1, 6, "", ("93.184.216.34", 0))]
26 success, latency = check_dns("example.com")
27 assert success is True
28 assert latency >= 0
29
30 def test_failed_resolution(self):
31 with patch("i2p_apps.setup.network_probe.socket.getaddrinfo") as mock:
32 mock.side_effect = socket.gaierror("Name resolution failed")
33 success, latency = check_dns("nonexistent.invalid")
34 assert success is False
35 assert latency == 0.0
36
37
38class TestProbeDns:
39 def test_all_domains_resolve(self):
40 with patch("i2p_apps.setup.network_probe.check_dns") as mock:
41 mock.return_value = (True, 10.0)
42 result = probe_dns()
43 assert isinstance(result, DnsProbeResult)
44 assert result.can_resolve_neutral is True
45 assert result.can_resolve_cdn is True
46 assert result.resolver_latency_ms > 0
47
48 def test_partial_resolution(self):
49 results = {
50 "example.com": (True, 10.0),
51 "cloudflare.com": (False, 0.0),
52 }
53 with patch("i2p_apps.setup.network_probe.check_dns") as mock:
54 mock.side_effect = lambda d: results.get(d, (False, 0.0))
55 result = probe_dns()
56 assert result.can_resolve_neutral is True
57 assert result.can_resolve_cdn is False
58
59
60class TestMeasureTcpRtt:
61 def test_successful_connection(self):
62 mock_sock = MagicMock()
63 mock_sock.connect_ex.return_value = 0
64 with patch("i2p_apps.setup.network_probe.socket.socket", return_value=mock_sock):
65 rtt = measure_tcp_rtt("example.com", 443)
66 assert rtt is not None
67 assert rtt >= 0
68
69 def test_connection_timeout(self):
70 mock_sock = MagicMock()
71 mock_sock.connect_ex.side_effect = OSError("timeout")
72 with patch("i2p_apps.setup.network_probe.socket.socket", return_value=mock_sock):
73 rtt = measure_tcp_rtt("unreachable.invalid", 443)
74 assert rtt is None
75
76
77class TestProbeLatency:
78 def test_returns_latency_result(self):
79 with patch("i2p_apps.setup.network_probe.measure_tcp_rtt") as mock:
80 mock.return_value = 0.045 # 45ms
81 result = probe_latency(endpoints=["example.com:443", "cloudflare.com:443"])
82 assert isinstance(result, LatencyProbeResult)
83 assert result.rtt_ms > 0
84 assert result.endpoints_reached > 0
85
86 def test_no_endpoints_reachable(self):
87 with patch("i2p_apps.setup.network_probe.measure_tcp_rtt", return_value=None):
88 result = probe_latency(endpoints=["unreachable.invalid:443"])
89 assert result.endpoints_reached == 0
90 assert result.rtt_ms == 0.0
91
92
93class TestEstimateBandwidth:
94 def test_successful_download(self):
95 mock_resp = MagicMock()
96 mock_resp.read.return_value = b"x" * 102400
97 mock_resp.status = 200
98 mock_resp.__enter__ = lambda s: s
99 mock_resp.__exit__ = MagicMock(return_value=False)
100
101 with patch("i2p_apps.setup.network_probe.urllib.request.urlopen",
102 return_value=mock_resp):
103 result = estimate_bandwidth(url="https://example.com/test")
104 assert result is not None
105 assert isinstance(result, BandwidthEstimate)
106 assert result.download_mbps > 0
107 assert result.method == "cdn_download"
108
109 def test_download_failure(self):
110 with patch("i2p_apps.setup.network_probe.urllib.request.urlopen",
111 side_effect=OSError("Connection refused")):
112 result = estimate_bandwidth(url="https://unreachable.invalid/test")
113 assert result is None
114
115
116class TestRunNetworkProbes:
117 def test_full_probe(self):
118 with patch("i2p_apps.setup.network_probe.probe_dns") as mock_dns, \
119 patch("i2p_apps.setup.network_probe.probe_latency") as mock_lat, \
120 patch("i2p_apps.setup.network_probe.estimate_bandwidth") as mock_bw, \
121 patch("i2p_apps.setup.network_probe.check_dns") as mock_check:
122 mock_dns.return_value = DnsProbeResult(
123 can_resolve_neutral=True, can_resolve_cdn=True,
124 can_resolve_reseed=True, resolver_latency_ms=10.0,
125 )
126 mock_lat.return_value = LatencyProbeResult(
127 rtt_ms=45.0, jitter_ms=8.0,
128 endpoints_reached=3, endpoints_tested=3,
129 )
130 mock_bw.return_value = BandwidthEstimate(
131 download_mbps=12.0, method="cdn_download",
132 )
133 mock_check.return_value = (True, 10.0)
134
135 result = run_network_probes()
136 assert isinstance(result, NetworkProbeResult)
137 assert result.dns.can_resolve_neutral is True
138 assert result.latency.rtt_ms == 45.0
139 assert result.bandwidth is not None
140
141 def test_no_bandwidth_probe(self):
142 with patch("i2p_apps.setup.network_probe.probe_dns") as mock_dns, \
143 patch("i2p_apps.setup.network_probe.probe_latency") as mock_lat, \
144 patch("i2p_apps.setup.network_probe.check_dns") as mock_check:
145 mock_dns.return_value = DnsProbeResult(
146 can_resolve_neutral=True, can_resolve_cdn=True,
147 can_resolve_reseed=False, resolver_latency_ms=10.0,
148 )
149 mock_lat.return_value = LatencyProbeResult(
150 rtt_ms=45.0, jitter_ms=8.0,
151 endpoints_reached=2, endpoints_tested=3,
152 )
153 mock_check.return_value = (True, 10.0)
154
155 result = run_network_probes(include_bandwidth=False)
156 assert result.bandwidth is None