proxy-port-monitor-tui/tests/test_proxy_monitor_tui.py

from __future__ import annotations

import importlib.util
import json
from pathlib import Path
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any
import unittest
import urllib.parse


def load_module():
    module_path = Path(__file__).resolve().parents[1] / "proxy_monitor_tui.py"
    spec = importlib.util.spec_from_file_location("proxy_monitor_tui", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


class ApiFixtureHandler(BaseHTTPRequestHandler):
    requests: list[dict[str, Any]] = []
    large_status = False

    def _json(self, status: int, payload: object) -> None:
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:  # noqa: N802
        parsed = urllib.parse.urlparse(self.path)
        type(self).requests.append({"method": "GET", "path": parsed.path, "query": urllib.parse.parse_qs(parsed.query)})
        if parsed.path == "/base/status.json":
            payload: dict[str, object] = {
                "targets_up": 1,
                "targets_total": 1,
                "targets": [{"name": "pc", "ok": True}],
            }
            if type(self).large_status:
                payload["padding"] = "x" * 1_100_000
            self._json(200, payload)
            return
        if parsed.path == "/base/admin/pc%20win/api/proxies":
            self._json(
                200,
                {
                    "proxies": {
                        "Selector A": {
                            "type": "Selector",
                            "now": "Node A",
                            "all": ["Node A", "Node B"],
                        }
                    }
                },
            )
            return
        if parsed.path == "/base/admin/pc%20win/api/proxies/Node%20A/delay":
            self._json(200, {"delay": 42})
            return
        self._json(404, {"error": "not found"})

    def do_PUT(self) -> None:  # noqa: N802
        parsed = urllib.parse.urlparse(self.path)
        length = int(self.headers.get("Content-Length", "0"))
        payload = json.loads(self.rfile.read(length).decode("utf-8"))
        type(self).requests.append({"method": "PUT", "path": parsed.path, "payload": payload})
        if parsed.path == "/base/admin/pc%20win/api/proxies/Selector%20A":
            self.send_response(204)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        self._json(404, {"error": "not found"})

    def log_message(self, format: str, *args: object) -> None:  # noqa: A003
        return


class ProxyMonitorTUITests(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), ApiFixtureHandler)
        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.thread.start()

    @classmethod
    def tearDownClass(cls) -> None:
        cls.server.shutdown()
        cls.thread.join(timeout=5)
        cls.server.server_close()

    def setUp(self) -> None:
        ApiFixtureHandler.requests = []
        ApiFixtureHandler.large_status = False
        self.mod = load_module()
        self.base_url = f"http://127.0.0.1:{self.server.server_port}/base"
        self.client = self.mod.ProxyMonitorClient(self.base_url, 2)

    def test_status_and_controller_paths_keep_the_base_prefix_and_quote_names(self) -> None:
        status = self.client.fetch_status()
        proxies = self.client.fetch_proxies("pc win")
        delay = self.client.measure_delay(
            "pc win",
            "Node A",
            test_url="https://www.gstatic.com/generate_204",
            timeout_ms=5000,
        )
        self.client.select_proxy("pc win", "Selector A", "Node B")

        self.assertEqual(status["targets_up"], 1)
        self.assertIn("Selector A", proxies["proxies"])
        self.assertEqual(delay, 42)
        self.assertEqual(
            ApiFixtureHandler.requests,
            [
                {"method": "GET", "path": "/base/status.json", "query": {}},
                {"method": "GET", "path": "/base/admin/pc%20win/api/proxies", "query": {}},
                {
                    "method": "GET",
                    "path": "/base/admin/pc%20win/api/proxies/Node%20A/delay",
                    "query": {"url": ["https://www.gstatic.com/generate_204"], "timeout": ["5000"]},
                },
                {
                    "method": "PUT",
                    "path": "/base/admin/pc%20win/api/proxies/Selector%20A",
                    "payload": {"name": "Node B"},
                },
            ],
        )

    def test_status_response_larger_than_one_mebibyte_is_not_truncated(self) -> None:
        ApiFixtureHandler.large_status = True

        status = self.client.fetch_status()

        self.assertEqual(len(status["padding"]), 1_100_000)

    def test_selector_groups_only_include_switchable_groups(self) -> None:
        groups = self.mod.normalize_selector_groups(
            {
                "proxies": {
                    "url-test": {"type": "URLTest", "all": ["A"]},
                    "selector": {"type": "Selector", "now": "A", "all": ["A", "B", "A"]},
                    "empty": {"type": "Selector", "now": "", "all": []},
                }
            }
        )
        self.assertEqual(groups, [{"name": "selector", "now": "A", "candidates": ["A", "B"]}])

    def test_recommendation_ignores_failed_or_zero_delay_candidates(self) -> None:
        recommended = self.mod.recommend_fastest(
            {
                "failure": (None, "timed out"),
                "zero": (0, "no delay result"),
                "slow": (340, "340 ms"),
                "fast": (81, "81 ms"),
            }
        )
        self.assertEqual(recommended, "fast")

    def test_machine_rows_preserve_monitor_order_and_status(self) -> None:
        rows = self.mod.normalize_machine_rows(
            {
                "targets": [
                    {"name": "first", "ok": True, "detail": "latency=1ms", "exit_country": "Japan"},
                    {"name": "second", "ok": False, "consecutive_failures": 3},
                ]
            }
        )
        self.assertEqual([row["name"] for row in rows], ["first", "second"])
        self.assertEqual(rows[0]["state"], "online")
        self.assertEqual(rows[1]["state"], "offline")
        self.assertEqual(rows[1]["consecutive_failures"], 3)

    def test_invalid_api_url_is_rejected_before_network_access(self) -> None:
        with self.assertRaises(ValueError):
            self.mod.ProxyMonitorClient("proxy.tailbeb9ad.ts.net", 2)