100 lines
3.3 KiB
Python
100 lines
3.3 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, normalize_proxy_url, sanitize_settings
|
|
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_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()
|