distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for config advisor — TDD: tests before implementation."""
2
3import pytest
4
5from i2p_apps.setup.local_probe import LocalProbeResult
6from i2p_apps.setup.stun_probe import NatProbeResult
7from i2p_apps.setup.network_probe import (
8 BandwidthEstimate,
9 DnsProbeResult,
10 LatencyProbeResult,
11 NetworkProbeResult,
12)
13from i2p_apps.setup.config_advisor import ConfigRecommendation, advise
14
15
16def _local(crypto=15000, ram=8_000_000_000, total_ram=16_000_000_000,
17 cpus=8, disk=200.0, disk_free=100_000_000_000):
18 return LocalProbeResult(
19 crypto_ops_per_sec=crypto,
20 available_ram_bytes=ram,
21 total_ram_bytes=total_ram,
22 cpu_count=cpus,
23 disk_write_mbps=disk,
24 disk_free_bytes=disk_free,
25 )
26
27
28def _nat(present=True, nat_type="cone", ip="203.0.113.42", port=54321):
29 return NatProbeResult(
30 nat_present=present,
31 nat_type=nat_type,
32 external_ip=ip,
33 external_port=port,
34 servers_reached=2,
35 )
36
37
38def _network(bw_mbps=12.0):
39 return NetworkProbeResult(
40 dns=DnsProbeResult(True, True, True, 10.0),
41 latency=LatencyProbeResult(45.0, 8.0, 3, 3),
42 bandwidth=BandwidthEstimate(bw_mbps, "cdn_download") if bw_mbps else None,
43 reseed_reachable=True,
44 )
45
46
47class TestAdvise:
48 def test_returns_config_recommendation(self):
49 result = advise(_local())
50 assert isinstance(result, ConfigRecommendation)
51
52 def test_high_cpu_gets_more_tunnels(self):
53 low = advise(_local(crypto=1000))
54 high = advise(_local(crypto=60000))
55 assert high.inbound_tunnel_count > low.inbound_tunnel_count
56
57 def test_symmetric_nat_disables_upnp(self):
58 result = advise(_local(), nat=_nat(nat_type="symmetric"))
59 assert result.upnp_enabled is False
60 assert any("ymmetric" in w for w in result.warnings)
61
62 def test_cone_nat_enables_upnp(self):
63 result = advise(_local(), nat=_nat(nat_type="cone"))
64 assert result.upnp_enabled is True
65
66 def test_open_internet_no_upnp(self):
67 result = advise(_local(), nat=_nat(present=False, nat_type="open"))
68 assert result.upnp_enabled is False
69
70 def test_high_bandwidth_high_share(self):
71 result = advise(_local(), network=_network(bw_mbps=100.0))
72 assert result.share_percentage >= 60
73 assert result.bandwidth_limit_kbps > 0
74
75 def test_low_bandwidth_low_share(self):
76 result = advise(_local(), network=_network(bw_mbps=0.5))
77 assert result.share_percentage <= 40
78
79 def test_no_network_uses_defaults(self):
80 result = advise(_local())
81 assert result.share_percentage == 50
82 assert result.bandwidth_limit_kbps == 0
83
84 def test_floodfill_requires_good_cpu_and_bandwidth(self):
85 # High CPU + high bandwidth + cone NAT = floodfill
86 result = advise(
87 _local(crypto=60000),
88 nat=_nat(nat_type="cone"),
89 network=_network(bw_mbps=100.0),
90 )
91 assert result.floodfill is True
92
93 def test_symmetric_nat_blocks_floodfill(self):
94 result = advise(
95 _local(crypto=60000),
96 nat=_nat(nat_type="symmetric"),
97 network=_network(bw_mbps=100.0),
98 )
99 assert result.floodfill is False
100
101 def test_low_cpu_blocks_floodfill(self):
102 result = advise(
103 _local(crypto=1000),
104 nat=_nat(nat_type="cone"),
105 network=_network(bw_mbps=100.0),
106 )
107 assert result.floodfill is False
108
109 def test_listen_port_default(self):
110 result = advise(_local())
111 assert isinstance(result.listen_port, int)
112 assert result.listen_port > 0
113
114 def test_to_router_config_overrides(self):
115 result = advise(_local(), nat=_nat(), network=_network())
116 overrides = result.to_router_config_overrides()
117 assert isinstance(overrides, dict)
118 assert "bandwidth_limit_kbps" in overrides
119 assert "share_percentage" in overrides