c9c45fdfec
- load_settings/sanitize_settings: treat sources=[] as invalid; restore defaults (all([]) was true) - _build_pool: defensive default sources + clearer empty-fetch log; elite-only empty fetch message - FirstRun_Build_And_Install.bat: Python check + setup_and_build.ps1 - README + test for empty sources sanitize Made-with: Cursor
134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
"""Automated checks for Proxy God core logic (no GUI). Run: python -m unittest discover -s tests -v"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import unittest
|
|
|
|
from proxy_chain_manager.config import (
|
|
Settings,
|
|
merge_proxy_credentials,
|
|
normalize_proxy_url,
|
|
redact_proxy_url,
|
|
sanitize_settings,
|
|
split_proxy_for_edit,
|
|
)
|
|
from proxy_chain_manager.fetcher import fetch_proxy_json, normalize_entries
|
|
from proxy_chain_manager.validator import (
|
|
check_chain_exit_ip,
|
|
get_direct_ip,
|
|
validate_proxies,
|
|
)
|
|
|
|
|
|
class TestConfig(unittest.TestCase):
|
|
def test_normalize_proxy_url(self) -> None:
|
|
self.assertEqual(normalize_proxy_url(""), "")
|
|
self.assertEqual(normalize_proxy_url("1.2.3.4:8080"), "http://1.2.3.4:8080")
|
|
self.assertEqual(normalize_proxy_url("socks5://x:1"), "socks5://x:1")
|
|
|
|
def test_merge_split_proxy_auth(self) -> None:
|
|
merged = merge_proxy_credentials("http://1.2.3.4:8080", "user", "p:ass")
|
|
self.assertIn("user", merged)
|
|
self.assertIn("p%3Aass", merged)
|
|
base, u, pw = split_proxy_for_edit(merged)
|
|
self.assertEqual(base, "http://1.2.3.4:8080")
|
|
self.assertEqual(u, "user")
|
|
self.assertEqual(pw, "p:ass")
|
|
|
|
def test_merge_preserves_embedded_auth_when_fields_empty(self) -> None:
|
|
raw = normalize_proxy_url("http://x:y@9.9.9.9:12")
|
|
self.assertEqual(merge_proxy_credentials(raw, "", ""), raw)
|
|
|
|
def test_redact_proxy_url(self) -> None:
|
|
r = redact_proxy_url("http://alice:secret@proxy.example:8888")
|
|
self.assertIn("***:***@", r)
|
|
self.assertIn("proxy.example", r)
|
|
self.assertNotIn("secret", r)
|
|
self.assertNotIn("alice", r)
|
|
|
|
def test_sanitize_restores_empty_sources(self) -> None:
|
|
s = Settings()
|
|
s.sources = []
|
|
s2, changed = sanitize_settings(s)
|
|
self.assertTrue(changed)
|
|
self.assertGreater(len(s2.sources), 0)
|
|
|
|
def test_sanitize_clamps_extremes(self) -> None:
|
|
s = Settings(
|
|
chain_length=0,
|
|
health_check_seconds=999999,
|
|
full_refresh_seconds=10,
|
|
validation_timeout_seconds=900.0,
|
|
)
|
|
s, changed = sanitize_settings(s)
|
|
self.assertTrue(changed)
|
|
self.assertEqual(s.chain_length, 1)
|
|
self.assertEqual(s.health_check_seconds, 3600)
|
|
self.assertEqual(s.full_refresh_seconds, 60)
|
|
self.assertEqual(s.validation_timeout_seconds, 120.0)
|
|
|
|
|
|
class TestFetcher(unittest.TestCase):
|
|
def test_normalize_entries_empty(self) -> None:
|
|
self.assertEqual(normalize_entries([], False), [])
|
|
|
|
def test_fetch_smoke(self) -> None:
|
|
"""One real HTTP GET — may fail offline; then skip assertion."""
|
|
try:
|
|
rows = fetch_proxy_json(
|
|
"https://cdn.jsdelivr.net/gh/proxifly/free-proxy-list@main/proxies/protocols/http/data.json",
|
|
timeout=25.0,
|
|
)
|
|
except Exception:
|
|
self.skipTest("network unavailable")
|
|
self.assertIsInstance(rows, list)
|
|
if rows:
|
|
self.assertTrue(isinstance(rows[0], dict))
|
|
|
|
|
|
class TestValidator(unittest.TestCase):
|
|
def test_validate_empty_list_returns_fast(self) -> None:
|
|
async def run() -> list[str]:
|
|
return await validate_proxies(
|
|
[],
|
|
"https://api.ipify.org?format=json",
|
|
8,
|
|
5.0,
|
|
)
|
|
|
|
self.assertEqual(asyncio.run(run()), [])
|
|
|
|
def test_validate_dead_proxy_fast(self) -> None:
|
|
async def run() -> list[str]:
|
|
return await validate_proxies(
|
|
["http://127.0.0.1:1"],
|
|
"https://api.ipify.org?format=json",
|
|
1,
|
|
3.0,
|
|
)
|
|
|
|
self.assertEqual(asyncio.run(run()), [])
|
|
|
|
def test_get_direct_ip_network(self) -> None:
|
|
async def run() -> str | None:
|
|
return await get_direct_ip("https://api.ipify.org?format=json", 12.0)
|
|
|
|
ip = asyncio.run(run())
|
|
if ip is None:
|
|
self.skipTest("could not reach ipify (offline or blocked)")
|
|
self.assertGreater(len(ip), 4)
|
|
|
|
def test_check_chain_exit_ip_invalid_proxy_quick(self) -> None:
|
|
async def run() -> str | None:
|
|
return await check_chain_exit_ip(
|
|
"http://127.0.0.1:1",
|
|
"https://api.ipify.org?format=json",
|
|
8.0,
|
|
)
|
|
|
|
self.assertIsNone(asyncio.run(run()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|