socks5-monitor/tests/test_proxy_validation.py

#!/usr/bin/env python3
"""Regression tests for node id generation, host normalization and error text.

These cover the three defects that made the panel look broken:
  1. Chinese node names slugged into ids that collided with existing nodes.
  2. A pasted URL in the Host field went straight to DNS and always failed.
  3. asyncio.TimeoutError has an empty str(), so the UI showed a blank error.
"""
from __future__ import annotations

import asyncio
import importlib.util
import json
import socket
import sys
import tempfile
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location("socks5_app", ROOT / "app.py")
assert spec is not None and spec.loader is not None
app = importlib.util.module_from_spec(spec)
sys.modules["socks5_app"] = app
spec.loader.exec_module(app)


def make_app(proxies: list[dict] | None = None) -> tuple[object, Path, tempfile.TemporaryDirectory]:
    """Build an App on a throwaway config; caller must use the TemporaryDirectory."""
    td = tempfile.TemporaryDirectory()
    cfg = Path(td.name) / "config.json"
    cfg.write_text(json.dumps({
        "proxies": proxies or [],
        "notifications": {"channels": []},
        "db_path": str(Path(td.name) / "monitor.db"),
    }), encoding="utf-8")
    return app.App(cfg), cfg, td


class AppCase(unittest.TestCase):
    """Base class that builds an App and always closes its SQLite handle."""

    def build(self, proxies: list[dict] | None = None):
        application, cfg, td = make_app(proxies)
        self.addCleanup(td.cleanup)
        self.addCleanup(application.store.close)
        return application, cfg


class NormalizeHostTest(unittest.TestCase):
    def test_bare_host_has_no_port(self):
        self.assertEqual(app.normalize_host("192.168.3.71"), ("192.168.3.71", None))

    def test_host_colon_port(self):
        self.assertEqual(app.normalize_host("192.168.3.71:7890"), ("192.168.3.71", 7890))

    def test_full_url_is_stripped(self):
        self.assertEqual(app.normalize_host("http://192.168.3.71:7890"), ("192.168.3.71", 7890))

    def test_scheme_and_credentials_are_stripped(self):
        self.assertEqual(app.normalize_host("socks5://user:pw@1.2.3.4:1080"), ("1.2.3.4", 1080))

    def test_trailing_path_is_dropped(self):
        self.assertEqual(app.normalize_host("socks5h://host.example.com:1080/"), ("host.example.com", 1080))

    def test_bracketed_ipv6(self):
        self.assertEqual(app.normalize_host("[2001:db8::1]:1080"), ("2001:db8::1", 1080))

    def test_bare_ipv6_is_not_split_on_colons(self):
        self.assertEqual(app.normalize_host("2001:db8::1"), ("2001:db8::1", None))

    def test_whitespace_and_empty(self):
        self.assertEqual(app.normalize_host("  chickliu.fun  "), ("chickliu.fun", None))
        self.assertEqual(app.normalize_host(""), ("", None))
        self.assertEqual(app.normalize_host(None), ("", None))

    def test_non_numeric_port_is_kept_as_host(self):
        self.assertEqual(app.normalize_host("host:notaport"), ("host", None))


class DescribeExcTest(unittest.TestCase):
    def test_timeout_is_never_blank(self):
        for exc in (asyncio.TimeoutError(), TimeoutError()):
            msg = app.describe_exc(exc)
            self.assertTrue(msg.strip())
            self.assertIn("超时", msg)

    def test_incomplete_read_is_labelled_handshake(self):
        self.assertIn("握手", app.describe_exc(asyncio.IncompleteReadError(b"", 2)))

    def test_refused_and_dns(self):
        self.assertIn("拒绝", app.describe_exc(ConnectionRefusedError()))
        self.assertIn("解析", app.describe_exc(socket.gaierror(-2, "Name does not resolve")))

    def test_message_is_preserved_for_other_errors(self):
        self.assertEqual(app.describe_exc(RuntimeError("rep=0x05")), "rep=0x05")

    def test_bare_exception_falls_back_to_class_name(self):
        self.assertEqual(app.describe_exc(RuntimeError()), "RuntimeError")


class SlugIdTest(AppCase):
    def test_chinese_name_does_not_collide_with_existing_id(self):
        application, _ = self.build([{"id": "01", "name": "15690", "host": "h", "port": 1}])
        self.assertEqual(application.slug_id("香港01"), "node-1")

    def test_repeated_non_ascii_names_get_distinct_ids(self):
        application, _ = self.build()
        first = application.validate_proxy({"name": "香港01", "host": "1.1.1.1"})
        application.config["proxies"].append(first)
        second = application.validate_proxy({"name": "香港01", "host": "1.1.1.1"})
        self.assertNotEqual(first["id"], second["id"])
        self.assertEqual([first["id"], second["id"]], ["node-1", "node-2"])

    def test_emoji_only_name_is_handled(self):
        application, _ = self.build()
        self.assertEqual(application.slug_id("🇯🇵日本"), "node-1")

    def test_ascii_name_keeps_readable_slug(self):
        application, _ = self.build()
        self.assertEqual(application.slug_id("HK Node 01"), "HK-Node-01")

    def test_ascii_collision_is_suffixed(self):
        application, _ = self.build([{"id": "hk", "name": "hk", "host": "h", "port": 1}])
        self.assertEqual(application.slug_id("hk"), "hk-2")


class ValidateProxyTest(AppCase):
    def test_pasted_url_is_normalized_into_host_and_port(self):
        application, _ = self.build()
        p = application.validate_proxy({"name": "u", "host": "http://192.168.3.71:7890"})
        self.assertEqual((p["host"], p["port"]), ("192.168.3.71", 7890))

    def test_explicit_port_wins_over_url_port(self):
        application, _ = self.build()
        p = application.validate_proxy({"name": "u", "host": "http://1.2.3.4:7890", "port": 9050})
        self.assertEqual(p["port"], 9050)

    def test_missing_host_raises_value_error(self):
        application, _ = self.build()
        with self.assertRaises(ValueError):
            application.validate_proxy({"name": "no-host"})
        with self.assertRaises(ValueError):
            application.validate_proxy({"name": "blank", "host": "   "})

    def test_out_of_range_port_raises_value_error(self):
        application, _ = self.build()
        with self.assertRaises(ValueError):
            application.validate_proxy({"name": "p", "host": "1.1.1.1", "port": 70000})

    def test_edit_preserves_existing_id(self):
        existing = {"id": "410-5585", "name": "old", "host": "1.1.1.1", "port": 1080}
        application, _ = self.build([existing])
        p = application.validate_proxy({"name": "renamed", "host": "2.2.2.2"}, existing=existing)
        self.assertEqual(p["id"], "410-5585")
        self.assertEqual(p["name"], "renamed")


class MigrateProxyHostsTest(AppCase):
    def test_url_host_is_healed_at_startup(self):
        application, _ = self.build([
            {"id": "a", "name": "a", "host": "http://192.168.3.71", "port": 7890},
            {"id": "b", "name": "b", "host": "192.168.3.46", "port": 7890},
        ])
        self.assertTrue(application._migrated)
        self.assertEqual(application.get_proxy("a")["host"], "192.168.3.71")
        self.assertEqual(application.get_proxy("a")["port"], 7890)
        self.assertEqual(application.get_proxy("b")["host"], "192.168.3.46")

    def test_url_port_adopted_when_port_is_default(self):
        application, _ = self.build([
            {"id": "a", "name": "a", "host": "http://10.0.0.5:1088", "port": 1080},
        ])
        self.assertEqual(application.get_proxy("a")["port"], 1088)

    def test_clean_config_is_not_marked_migrated(self):
        application, _ = self.build([{"id": "a", "name": "a", "host": "1.1.1.1", "port": 1080}])
        self.assertFalse(application._migrated)

    def test_migration_is_persisted_by_start_tasks(self):
        application, cfg = self.build([
            {"id": "a", "name": "a", "host": "http://192.168.3.71", "port": 7890, "enabled": False},
        ])
        asyncio.run(application.start_tasks())
        saved = json.loads(cfg.read_text(encoding="utf-8"))
        self.assertEqual(saved["proxies"][0]["host"], "192.168.3.71")


class ManualCheckTest(AppCase):
    def test_force_runs_on_disabled_node_and_records_error(self):
        application, _ = self.build([
            {"id": "dead", "name": "dead", "host": "127.0.0.1", "port": 19999,
             "enabled": False, "timeout_seconds": 2},
        ])
        proxy = application.get_proxy("dead")
        asyncio.run(application.run_once(proxy, force=True))
        st = application.state_for("dead")
        self.assertIs(st.last_ok, False)
        self.assertTrue((st.error or "").strip(), "manual check must report a reason")

    def test_without_force_a_disabled_node_is_skipped(self):
        application, _ = self.build([
            {"id": "dead", "name": "dead", "host": "127.0.0.1", "port": 19999,
             "enabled": False, "timeout_seconds": 2},
        ])
        asyncio.run(application.run_once(application.get_proxy("dead")))
        self.assertIsNone(application.state_for("dead").last_ok)


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