#!/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()