5418b30441
Cover valid entries, invalid proto/port, missing keys, and mixed valid/invalid input. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
144 lines
4.6 KiB
Python
144 lines
4.6 KiB
Python
"""Tests for configuration loading and proxy URL parsing."""
|
|
|
|
import pytest
|
|
|
|
from s5p.config import ChainHop, Config, load_config, parse_api_proxies, parse_proxy_url
|
|
|
|
|
|
class TestParseProxyUrl:
|
|
"""Test proxy URL parsing."""
|
|
|
|
def test_socks5_basic(self):
|
|
hop = parse_proxy_url("socks5://127.0.0.1:9050")
|
|
assert hop.proto == "socks5"
|
|
assert hop.host == "127.0.0.1"
|
|
assert hop.port == 9050
|
|
assert hop.username is None
|
|
assert hop.password is None
|
|
|
|
def test_socks5_with_auth(self):
|
|
hop = parse_proxy_url("socks5://user:pass@proxy.example.com:1080")
|
|
assert hop.proto == "socks5"
|
|
assert hop.host == "proxy.example.com"
|
|
assert hop.port == 1080
|
|
assert hop.username == "user"
|
|
assert hop.password == "pass"
|
|
|
|
def test_socks4(self):
|
|
hop = parse_proxy_url("socks4://10.0.0.1:1080")
|
|
assert hop.proto == "socks4"
|
|
assert hop.host == "10.0.0.1"
|
|
assert hop.port == 1080
|
|
|
|
def test_http_connect(self):
|
|
hop = parse_proxy_url("http://proxy:8080")
|
|
assert hop.proto == "http"
|
|
assert hop.host == "proxy"
|
|
assert hop.port == 8080
|
|
|
|
def test_default_port_socks5(self):
|
|
hop = parse_proxy_url("socks5://host")
|
|
assert hop.port == 1080
|
|
|
|
def test_default_port_http(self):
|
|
hop = parse_proxy_url("http://host")
|
|
assert hop.port == 8080
|
|
|
|
def test_unsupported_protocol(self):
|
|
with pytest.raises(ValueError, match="unsupported protocol"):
|
|
parse_proxy_url("ftp://host:21")
|
|
|
|
def test_missing_host(self):
|
|
with pytest.raises(ValueError, match="missing host"):
|
|
parse_proxy_url("socks5://")
|
|
|
|
|
|
class TestChainHop:
|
|
"""Test ChainHop string representation."""
|
|
|
|
def test_str_without_auth(self):
|
|
hop = ChainHop(proto="socks5", host="localhost", port=9050)
|
|
assert str(hop) == "socks5://localhost:9050"
|
|
|
|
def test_str_with_auth(self):
|
|
hop = ChainHop(proto="http", host="proxy", port=8080, username="u", password="p")
|
|
assert str(hop) == "http://u@proxy:8080"
|
|
|
|
|
|
class TestParseApiProxies:
|
|
"""Test API response proxy parsing."""
|
|
|
|
def test_valid_entries(self):
|
|
data = {
|
|
"proxies": [
|
|
{"proto": "socks5", "proxy": "1.2.3.4:1080"},
|
|
{"proto": "http", "proxy": "5.6.7.8:8080"},
|
|
],
|
|
}
|
|
result = parse_api_proxies(data)
|
|
assert len(result) == 2
|
|
assert result[0] == ChainHop(proto="socks5", host="1.2.3.4", port=1080)
|
|
assert result[1] == ChainHop(proto="http", host="5.6.7.8", port=8080)
|
|
|
|
def test_skips_invalid_proto(self):
|
|
data = {"proxies": [{"proto": "ftp", "proxy": "1.2.3.4:21"}]}
|
|
assert parse_api_proxies(data) == []
|
|
|
|
def test_skips_missing_proto(self):
|
|
data = {"proxies": [{"proxy": "1.2.3.4:1080"}]}
|
|
assert parse_api_proxies(data) == []
|
|
|
|
def test_skips_missing_colon(self):
|
|
data = {"proxies": [{"proto": "socks5", "proxy": "no-port"}]}
|
|
assert parse_api_proxies(data) == []
|
|
|
|
def test_skips_bad_port(self):
|
|
data = {"proxies": [{"proto": "socks5", "proxy": "1.2.3.4:abc"}]}
|
|
assert parse_api_proxies(data) == []
|
|
|
|
def test_empty_proxies(self):
|
|
assert parse_api_proxies({"proxies": []}) == []
|
|
|
|
def test_missing_proxies_key(self):
|
|
assert parse_api_proxies({}) == []
|
|
|
|
def test_mixed_valid_invalid(self):
|
|
data = {
|
|
"proxies": [
|
|
{"proto": "socks5", "proxy": "1.2.3.4:1080"},
|
|
{"proto": "ftp", "proxy": "bad:21"},
|
|
{"proto": "socks4", "proxy": "5.6.7.8:1080"},
|
|
],
|
|
}
|
|
result = parse_api_proxies(data)
|
|
assert len(result) == 2
|
|
assert result[0].proto == "socks5"
|
|
assert result[1].proto == "socks4"
|
|
|
|
|
|
class TestConfig:
|
|
"""Test Config defaults."""
|
|
|
|
def test_defaults(self):
|
|
c = Config()
|
|
assert c.listen_host == "127.0.0.1"
|
|
assert c.listen_port == 1080
|
|
assert c.chain == []
|
|
assert c.timeout == 10.0
|
|
assert c.max_connections == 256
|
|
assert c.pool_size == 0
|
|
assert c.pool_max_idle == 30.0
|
|
|
|
def test_max_connections_from_yaml(self, tmp_path):
|
|
cfg_file = tmp_path / "test.yaml"
|
|
cfg_file.write_text("max_connections: 512\n")
|
|
c = load_config(cfg_file)
|
|
assert c.max_connections == 512
|
|
|
|
def test_pool_size_from_yaml(self, tmp_path):
|
|
cfg_file = tmp_path / "test.yaml"
|
|
cfg_file.write_text("pool_size: 16\npool_max_idle: 45.0\n")
|
|
c = load_config(cfg_file)
|
|
assert c.pool_size == 16
|
|
assert c.pool_max_idle == 45.0
|