proxy-god/tests/test_proxy_god.py

"""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.leak_detect import is_chain_leak, is_same_subnet

from proxy_chain_manager.config import (
    Settings,
    merge_proxy_credentials,
    normalize_proxy_url,
    parse_exit_proxy_input,
    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,
    check_https_tunnel,
    get_direct_ip,
    validate_proxies,
)


class TestSameNetwork(unittest.TestCase):
    def test_exact_match(self) -> None:
        self.assertTrue(is_same_subnet("1.2.3.4", "1.2.3.4"))

    def test_same_slash16(self) -> None:
        self.assertTrue(is_same_subnet("185.220.101.5", "185.220.200.9"))

    def test_different_slash16(self) -> None:
        self.assertFalse(is_same_subnet("185.220.101.5", "104.28.3.99"))

    def test_none_safe(self) -> None:
        self.assertFalse(is_same_subnet(None, "1.2.3.4"))
        self.assertFalse(is_same_subnet("1.2.3.4", None))

    def test_non_ipv4_safe(self) -> None:
        self.assertFalse(is_same_subnet("not-an-ip", "1.2.3.4"))


class TestLeakDetect(unittest.TestCase):
    def test_strict_no_vpn_same_ip(self) -> None:
        self.assertTrue(is_chain_leak("1.2.3.4", "1.2.3.4", vpn_active=False))

    def test_strict_no_vpn_different_ip(self) -> None:
        self.assertFalse(is_chain_leak("5.6.7.8", "1.2.3.4", vpn_active=False))

    def test_vpn_subnet_leak(self) -> None:
        self.assertTrue(is_chain_leak("185.220.101.5", "185.220.200.9", vpn_active=True))

    def test_vpn_different_subnet_ok(self) -> None:
        self.assertFalse(is_chain_leak("104.28.3.99", "185.220.101.5", vpn_active=True))


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_parse_exit_host_port_only_defaults_to_socks5(self) -> None:
        base, u, pw = parse_exit_proxy_input("1.2.3.4:1080")
        self.assertEqual(base, "socks5://1.2.3.4:1080")
        self.assertEqual(u, "")
        self.assertEqual(pw, "")

    def test_parse_exit_colon_quadruple(self) -> None:
        base, u, pw = parse_exit_proxy_input("1.2.3.4:1080:alice:s3cret")
        self.assertEqual(base, "socks5://1.2.3.4:1080")
        self.assertEqual(u, "alice")
        self.assertEqual(pw, "s3cret")

    def test_parse_exit_password_with_colon_kept_intact(self) -> None:
        base, u, pw = parse_exit_proxy_input("h:9:u:a:b:c")
        self.assertEqual(base, "socks5://h:9")
        self.assertEqual(u, "u")
        self.assertEqual(pw, "a:b:c")

    def test_parse_exit_at_form_no_scheme(self) -> None:
        base, u, pw = parse_exit_proxy_input("alice:secret@1.2.3.4:1080")
        self.assertEqual(base, "socks5://1.2.3.4:1080")
        self.assertEqual(u, "alice")
        self.assertEqual(pw, "secret")

    def test_parse_exit_respects_explicit_scheme(self) -> None:
        base, u, pw = parse_exit_proxy_input("http://x:y@9.9.9.9:8080")
        self.assertEqual(base, "http://9.9.9.9:8080")
        self.assertEqual(u, "x")
        self.assertEqual(pw, "y")

    def test_parse_exit_whitespace_form(self) -> None:
        base, u, pw = parse_exit_proxy_input("1.2.3.4:1080 alice s3cret")
        self.assertEqual(base, "socks5://1.2.3.4:1080")
        self.assertEqual(u, "alice")
        self.assertEqual(pw, "s3cret")

    def test_parse_exit_empty(self) -> None:
        self.assertEqual(parse_exit_proxy_input("   "), ("", "", ""))

    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_validate_early_exit_target(self) -> None:
        """Once target is reached remaining tasks should be skipped (all dead here → no early exit but no crash)."""
        async def run() -> list[str]:
            return await validate_proxies(
                ["http://127.0.0.1:1"] * 10,
                "https://api.ipify.org?format=json",
                4,
                2.0,
                target=2,   # target unreachable — should still return [] cleanly
            )

        result = asyncio.run(run())
        self.assertEqual(result, [])

    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_https_tunnel_invalid_proxy_quick(self) -> None:
        async def run() -> tuple[bool, str]:
            return await check_https_tunnel("http://127.0.0.1:1", timeout_seconds=4.0)

        ok, msg = asyncio.run(run())
        self.assertFalse(ok)
        self.assertTrue(msg)

    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()))


class TestBrowserPersona(unittest.TestCase):
    def test_blend_persona_relaxes_block_flags(self) -> None:
        from proxy_chain_manager.browser_profile import (
            FirefoxHardening, apply_persona, build_user_js,
        )
        h = FirefoxHardening(
            persona_key="blend_windows_chrome",
            resist_fingerprinting=True,
            first_party_isolation=True,
            strict_tracking_protection=True,
            timezone_utc=True,
        )
        applied = apply_persona(h)
        self.assertFalse(applied.resist_fingerprinting)
        self.assertFalse(applied.first_party_isolation)
        self.assertFalse(applied.strict_tracking_protection)
        self.assertFalse(applied.timezone_utc)
        self.assertIsNotNone(applied.persona)
        # And the user.js carries a spoofed UA when blend is selected.
        js = build_user_js("127.0.0.1", 18888, h)
        self.assertIn("general.useragent.override", js)
        self.assertIn("Chrome/124", js)

    def test_hardened_persona_keeps_block_flags(self) -> None:
        from proxy_chain_manager.browser_profile import FirefoxHardening, apply_persona
        h = FirefoxHardening(
            persona_key="hardened",
            resist_fingerprinting=False,  # should be forced True
            first_party_isolation=False,
        )
        applied = apply_persona(h)
        self.assertTrue(applied.resist_fingerprinting)
        self.assertTrue(applied.first_party_isolation)
        self.assertTrue(applied.strict_tracking_protection)
        self.assertIsNone(applied.persona)

    def test_cookie_behavior_mapping(self) -> None:
        from proxy_chain_manager.browser_identity import (
            cookie_behavior_value, cookie_session_only,
        )
        self.assertEqual(cookie_behavior_value("accept_all"), 0)
        self.assertEqual(cookie_behavior_value("block_third_party"), 1)
        self.assertEqual(cookie_behavior_value("block_third_party_trackers"), 4)
        self.assertEqual(cookie_behavior_value("session_only"), 0)
        self.assertEqual(cookie_behavior_value("block_all"), 2)
        self.assertTrue(cookie_session_only("session_only"))
        self.assertFalse(cookie_session_only("accept_all"))

    def test_persona_env_returns_tz_for_blend(self) -> None:
        from proxy_chain_manager.browser_profile import FirefoxHardening, persona_env
        env = persona_env(FirefoxHardening(persona_key="blend_windows_chrome"))
        self.assertIn("TZ", env)
        self.assertTrue(env["TZ"])

    def test_persona_env_empty_for_hardened(self) -> None:
        from proxy_chain_manager.browser_profile import FirefoxHardening, persona_env
        self.assertEqual(persona_env(FirefoxHardening(persona_key="hardened")), {})


class TestPrivacyLan(unittest.TestCase):
    def test_snapshot_round_trip(self) -> None:
        from proxy_chain_manager.privacy_lan import (
            LanSnapshot, snapshot_from_json, snapshot_to_json,
        )
        s = LanSnapshot(
            netbios_per_interface={"{a-b-c}": 0, "{d-e-f}": 2},
            llmnr_present=True,
            llmnr_prev_value=1,
            mdns_present=False,
            mdns_prev_value=None,
        )
        round_tripped = snapshot_from_json(snapshot_to_json(s))
        self.assertIsNotNone(round_tripped)
        assert round_tripped is not None
        self.assertEqual(round_tripped.netbios_per_interface, s.netbios_per_interface)
        self.assertEqual(round_tripped.llmnr_present, s.llmnr_present)
        self.assertEqual(round_tripped.llmnr_prev_value, s.llmnr_prev_value)
        self.assertEqual(round_tripped.mdns_present, s.mdns_present)
        self.assertEqual(round_tripped.mdns_prev_value, s.mdns_prev_value)

    def test_lan_status_returns_keys(self) -> None:
        from proxy_chain_manager.privacy_lan import lan_status
        st = lan_status()
        self.assertIn("netbios", st)
        self.assertIn("llmnr", st)
        self.assertIn("mdns", st)


class TestLeakAuditPure(unittest.TestCase):
    def test_categorize_exit_private(self) -> None:
        from proxy_chain_manager.leak_audit import _categorize_exit
        self.assertEqual(_categorize_exit("10.0.0.5"), "PRIVATE — chain broken")
        self.assertEqual(_categorize_exit("192.168.1.1"), "PRIVATE — chain broken")
        self.assertEqual(_categorize_exit("172.20.5.5"), "PRIVATE — chain broken")

    def test_categorize_exit_loopback(self) -> None:
        from proxy_chain_manager.leak_audit import _categorize_exit
        self.assertEqual(_categorize_exit("127.0.0.1"), "LOOPBACK — chain broken")

    def test_categorize_exit_public(self) -> None:
        from proxy_chain_manager.leak_audit import _categorize_exit
        self.assertEqual(_categorize_exit("104.28.3.99"), "public")
        self.assertEqual(_categorize_exit("8.8.8.8"), "public")

    def test_categorize_exit_none(self) -> None:
        from proxy_chain_manager.leak_audit import _categorize_exit
        self.assertEqual(_categorize_exit(None), "unknown")

    def test_audit_finding_aggregation(self) -> None:
        from proxy_chain_manager.leak_audit import AuditReport
        rep = AuditReport()
        rep.add("X", True, "ok")
        self.assertTrue(rep.overall_ok)
        rep.add("Y", False, "bad", "note")
        self.assertFalse(rep.overall_ok)
        self.assertEqual(len(rep.findings), 2)


class TestExitIntelGeo(unittest.TestCase):
    def test_parse_ipapi_lat_lon(self) -> None:
        from proxy_chain_manager.exit_intel import _parse_ipapi

        out = _parse_ipapi(
            {
                "status": "success",
                "query": "8.8.8.8",
                "country": "United States",
                "countryCode": "US",
                "regionName": "Virginia",
                "city": "Ashburn",
                "lat": 39.03,
                "lon": -77.5,
                "timezone": "America/New_York",
                "isp": "Google LLC",
                "org": "Google Public DNS",
                "as": "AS15169 Google LLC",
                "mobile": False,
                "proxy": False,
                "hosting": True,
            }
        )
        self.assertTrue(out.ok)
        self.assertAlmostEqual(out.lat or 0, 39.03)
        self.assertAlmostEqual(out.lon or 0, -77.5)

    def test_parse_ipwhois_lat_lon(self) -> None:
        from proxy_chain_manager.exit_intel import _parse_ipwhois

        out = _parse_ipwhois(
            {
                "success": True,
                "ip": "1.1.1.1",
                "country": "Australia",
                "country_code": "AU",
                "city": "Sydney",
                "region": "New South Wales",
                "latitude": -33.86,
                "longitude": 151.2,
                "timezone": {"id": "Australia/Sydney"},
                "connection": {"asn": 13335, "org": "Cloudflare", "isp": "Cloudflare"},
            }
        )
        self.assertTrue(out.ok)
        self.assertAlmostEqual(out.lat or 0, -33.86)
        self.assertAlmostEqual(out.lon or 0, 151.2)


class TestChainMap(unittest.TestCase):
    def test_extract_hop_host(self) -> None:
        from proxy_chain_manager.chain_map import extract_hop_host

        self.assertEqual(extract_hop_host("http://203.0.113.10:8080"), "203.0.113.10")
        self.assertEqual(extract_hop_host("socks5://proxy.example:1080"), "proxy.example")
        self.assertIsNone(extract_hop_host(""))

    def test_project_corners(self) -> None:
        from proxy_chain_manager.chain_map import _project

        x, y = _project(0, 0, 360, 180, 0)
        self.assertAlmostEqual(x, 180.0)
        self.assertAlmostEqual(y, 90.0)

    def test_render_chain_map_with_points(self) -> None:
        from proxy_chain_manager.chain_map import MapPoint, render_chain_map

        pts = [
            MapPoint("you", "YOU", 40.0, -74.0, detail="NYC"),
            MapPoint("hop", "H1", 51.5, -0.1, detail="London"),
            MapPoint("exit", "EXIT", 35.6, 139.7, detail="Tokyo"),
        ]
        img = render_chain_map(pts, width=640, height=200, status="healthy")
        self.assertEqual(img.size, (640, 200))


if __name__ == "__main__":
    unittest.main()