dotfiles/tests/test_proxy_client.py

#!/usr/bin/env python3
import fcntl
import http.client
import json
import hashlib
import os
import runpy
import subprocess
import sys
import tempfile
import unittest
import urllib.error
import zipfile
from unittest import mock
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
CLI = REPO_ROOT / "proxy-client/.local/bin/proxy-client"


class FakeUrlResponse:
    def __init__(self, data=b"", content_length=None, read_error=None):
        self.data = data
        self.position = 0
        self.read_error = read_error
        self.headers = {}
        if content_length is not None:
            self.headers["Content-Length"] = str(content_length)

    def __enter__(self):
        return self

    def __exit__(self, _type, _value, _traceback):
        return False

    def read(self, size=-1):
        if self.read_error is not None:
            raise self.read_error
        if size < 0:
            size = len(self.data) - self.position
        chunk = self.data[self.position : self.position + size]
        self.position += len(chunk)
        return chunk


def checking_socket_setup(outcome="success"):
    if outcome == "success":
        outcome_line = "    return Connection()\n"
    elif outcome == "refused":
        outcome_line = "    raise OSError('connection refused')\n"
    else:
        raise ValueError("unknown socket outcome: {}".format(outcome))
    return (
        "import json, os, socket\n"
        "expected_connection = json.loads(os.environ['TEST_EXPECTED_CONNECTION'])\n"
        "class Connection:\n"
        "    def close(self):\n"
        "        pass\n"
        "def checked_create_connection(endpoint, timeout):\n"
        "    actual_connection = [endpoint[0], endpoint[1], timeout]\n"
        "    if actual_connection != expected_connection:\n"
        "        raise AssertionError('socket call mismatch: expected %r, got %r' "
        "% (expected_connection, actual_connection))\n"
        + outcome_line
        + "socket.create_connection = checked_create_connection"
    )


SUCCESSFUL_SOCKET_SETUP = checking_socket_setup()
REFUSED_SOCKET_SETUP = checking_socket_setup("refused")


class ProxyClientCliTest(unittest.TestCase):
    def setUp(self):
        self.temp_dir = tempfile.TemporaryDirectory()
        self.config_home = Path(self.temp_dir.name)
        self.environment = os.environ.copy()
        self.environment["XDG_CONFIG_HOME"] = str(self.config_home)
        self.environment["PYTHONDONTWRITEBYTECODE"] = "1"

    def tearDown(self):
        self.temp_dir.cleanup()

    def write_config(
        self,
        port,
        proxy_type="external",
        name="Test proxy",
        address="127.0.0.1",
        no_proxy_extra=".example.test",
    ):
        config_dir = self.config_home / "proxy-client"
        config_dir.mkdir(exist_ok=True)
        (config_dir / "config").write_text(
            "[proxy]\n"
            "type = {}\n"
            "name = {}\n"
            "address = {}\n"
            "port = {}\n"
            "no_proxy_extra = {}\n".format(
                proxy_type, name, address, port, no_proxy_extra
            ),
            encoding="utf-8",
        )

    def write_raw_config(self, content):
        config_dir = self.config_home / "proxy-client"
        config_dir.mkdir(exist_ok=True)
        path = config_dir / "config"
        if isinstance(content, bytes):
            path.write_bytes(content)
        else:
            path.write_text(content, encoding="utf-8")

    def run_cli(self, *arguments, environment=None):
        command_environment = self.environment.copy()
        if environment:
            command_environment.update(environment)
        return subprocess.run(
            [str(CLI)] + list(arguments),
            text=True,
            capture_output=True,
            env=command_environment,
            check=False,
        )

    def run_cli_with_socket_setup(
        self, arguments, setup, environment=None, expected_connection=None
    ):
        command_environment = self.environment.copy()
        if environment:
            command_environment.update(environment)
        command_environment["TEST_CLI_ARGUMENTS"] = json.dumps(list(arguments))
        if expected_connection is not None:
            command_environment["TEST_EXPECTED_CONNECTION"] = json.dumps(
                list(expected_connection)
            )
        code = (
            "import json, os, runpy, sys\n"
            + setup
            + "\nsys.argv = [sys.argv[1]] + "
            "json.loads(os.environ['TEST_CLI_ARGUMENTS'])\n"
            "runpy.run_path(sys.argv[0], run_name='__main__')"
        )
        return subprocess.run(
            [sys.executable, "-c", code, str(CLI)],
            text=True,
            capture_output=True,
            env=command_environment,
            check=False,
        )

    def test_socket_fixture_rejects_wrong_endpoint_and_timeout(self):
        self.write_config(23456)
        result = self.run_cli_with_socket_setup(
            ("status",),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("wrong.example", 1, 0.5),
        )

        self.assertNotEqual(0, result.returncode)
        self.assertIn("AssertionError: socket call mismatch", result.stderr)

    def test_no_subcommand_has_no_side_effect_and_reports_usage(self):
        result = self.run_cli()
        self.assertEqual(2, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("укажите явную подкоманду", result.stderr)
        self.assertFalse((self.config_home / "proxy-client").exists())

    def test_old_python_version_gets_clear_error_before_main_program(self):
        code = (
            "import runpy, sys; "
            "sys.version_info = (3, 7, 9); "
            "runpy.run_path(sys.argv[1], run_name='__main__')"
        )
        result = subprocess.run(
            [sys.executable, "-c", code, str(CLI)],
            text=True,
            capture_output=True,
            check=False,
        )
        self.assertEqual(1, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("требует Python 3.8", result.stderr)
        self.assertIn("Python 3.7", result.stderr)

    def test_proc_stat_parser_handles_spaces_and_closing_parentheses_in_comm(self):
        module = runpy.run_path(str(CLI))
        parse_starttime = module["parse_proc_stat_starttime"]
        tail = ["S"] + [str(value) for value in range(4, 22)] + ["987654"]

        for comm in ("worker", "worker name", "worker name ) with parens"):
            with self.subTest(comm=comm):
                self.assertEqual(
                    987654,
                    parse_starttime("321 ({}) ".format(comm) + " ".join(tail)),
                )
        for malformed in ("321 worker S 1 2", "321 (worker) S 1 2"):
            with self.subTest(malformed=malformed):
                with self.assertRaises(ValueError):
                    parse_starttime(malformed)

    def test_process_permission_error_is_not_treated_as_dead(self):
        module = runpy.run_path(str(CLI))
        with mock.patch.object(module["Path"], "stat", side_effect=PermissionError()):
            self.assertTrue(module["process_alive"](12345))

    def test_same_pid_cannot_be_a_healthy_managed_pair(self):
        module = runpy.run_path(str(CLI))
        function_globals = module["managed_state_outcome"].__globals__
        state = {"supervisor_pid": 12345, "xray_pid": 12345, "boot_id": "boot"}
        replacements = {
            "read_boot_id": mock.Mock(return_value="boot"),
            "process_alive": mock.Mock(return_value=True),
            "supervisor_identity_matches": mock.Mock(return_value=True),
            "xray_identity_matches": mock.Mock(return_value=True),
        }
        with mock.patch.dict(function_globals, replacements):
            self.assertEqual("unverifiable", module["managed_state_outcome"](state))

    def test_identity_reads_starttime_after_process_paths(self):
        module = runpy.run_path(str(CLI))
        function_globals = module["xray_identity_matches"].__globals__
        state = {
            "supervisor_pid": 101,
            "supervisor_starttime": 201,
            "supervisor_path": "/cli",
            "xray_pid": 102,
            "xray_starttime": 202,
            "xray_path": "/xray",
        }

        xray_events = []
        xray_replacements = {
            "process_executable": lambda _pid: xray_events.append("exe") or "/xray",
            "process_starttime": lambda _pid: xray_events.append("stat") or 202,
        }
        with mock.patch.dict(function_globals, xray_replacements):
            self.assertTrue(module["xray_identity_matches"](state))
        self.assertEqual(["exe", "stat"], xray_events)

        supervisor_events = []
        supervisor_replacements = {
            "process_cmdline": lambda _pid: supervisor_events.append("cmdline")
            or ["python", "/cli"],
            "process_argument_resolves_to": lambda *_args: supervisor_events.append(
                "script"
            )
            or True,
            "process_starttime": lambda _pid: supervisor_events.append("stat") or 201,
        }
        with mock.patch.dict(function_globals, supervisor_replacements):
            self.assertTrue(module["supervisor_identity_matches"](state))
        self.assertEqual(["cmdline", "script", "stat"], supervisor_events)

    def test_initial_child_starttime_retries_while_child_is_alive(self):
        module = runpy.run_path(str(CLI))
        wait_for_starttime = module["wait_for_child_starttime"]
        function_globals = wait_for_starttime.__globals__
        child = mock.Mock(pid=12345)
        child.poll.return_value = None
        replacements = {
            "process_starttime": mock.Mock(side_effect=[None, 7654321]),
            "time": mock.Mock(),
        }
        with mock.patch.dict(function_globals, replacements):
            self.assertEqual(7654321, wait_for_starttime(child))

    def test_broken_config_symlink_is_not_treated_as_missing_config(self):
        config_dir = self.config_home / "proxy-client"
        config_dir.mkdir()
        (config_dir / "config").symlink_to(self.config_home / "missing-config")
        result = self.run_cli("start")
        self.assertNotEqual(0, result.returncode)
        self.assertIn("отсутствующий файл", result.stderr)
        self.assertNotIn("только для type=xray", result.stderr)

    def test_missing_config_keeps_legacy_external_hiddify_endpoint(self):
        result = self.run_cli("status", environment={"PROXY_PREFLIGHT_TIMEOUT": "0.05"})
        output = result.stdout + result.stderr
        self.assertIn("Hiddify", output)
        self.assertIn("hproxy.local:12334", output)

    def test_status_reports_reachable_external_proxy(self):
        self.write_config(23456)
        result = self.run_cli_with_socket_setup(
            ("status",),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("127.0.0.1", 23456, 2.0),
        )
        self.assertEqual(0, result.returncode, result.stderr)
        self.assertIn("Test proxy (external) доступен", result.stdout)
        self.assertEqual("", result.stderr)

    def test_env_prints_complete_proxy_environment_for_external_proxy(self):
        self.write_config(
            23456, no_proxy_extra=" .example.test , git.example.test "
        )
        result = self.run_cli_with_socket_setup(
            ("env",),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("127.0.0.1", 23456, 2.0),
        )

        proxy_url = "http://127.0.0.1:23456"
        socks_url = "socks5://127.0.0.1:23456"
        no_proxy = (
            "127.0.0.1,localhost,::1,[::1],.example.test,git.example.test"
        )
        self.assertEqual(0, result.returncode, result.stderr)
        self.assertEqual("", result.stderr)
        self.assertEqual(
            [
                "HTTP_PROXY=" + proxy_url,
                "HTTPS_PROXY=" + proxy_url,
                "ALL_PROXY=" + socks_url,
                "NO_PROXY=" + no_proxy,
                "http_proxy=" + proxy_url,
                "https_proxy=" + proxy_url,
                "all_proxy=" + socks_url,
                "no_proxy=" + no_proxy,
            ],
            result.stdout.splitlines(),
        )

    def test_env_brackets_ipv6_without_requiring_host_ipv6(self):
        self.write_config(23456, address="2001:db8::1", no_proxy_extra="")
        result = self.run_cli_with_socket_setup(
            ("env",),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("2001:db8::1", 23456, 2.0),
        )

        self.assertEqual(0, result.returncode, result.stderr)
        self.assertIn("HTTP_PROXY=http://[2001:db8::1]:23456\n", result.stdout)
        self.assertIn("ALL_PROXY=socks5://[2001:db8::1]:23456\n", result.stdout)

    def test_scoped_ipv6_is_rejected_when_ipaddress_lacks_scope_support(self):
        self.write_config(23456, address="fe80::1%eth0", no_proxy_extra="")
        setup = (
            "import ipaddress\n"
            "real_ip_address = ipaddress.ip_address\n"
            "def python38_ip_address(value):\n"
            "    if '%' in value:\n"
            "        raise ValueError('scope IDs unsupported')\n"
            "    return real_ip_address(value)\n"
            "ipaddress.ip_address = python38_ip_address\n"
            + SUCCESSFUL_SOCKET_SETUP
        )
        result = self.run_cli_with_socket_setup(
            ("env",),
            setup,
            expected_connection=("fe80::1%eth0", 23456, 2.0),
        )

        self.assertNotEqual(0, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("address", result.stderr)

    def test_unreachable_env_prints_no_partial_output(self):
        self.write_config(23456)
        result = self.run_cli_with_socket_setup(
            ("env",),
            REFUSED_SOCKET_SETUP,
            {"PROXY_PREFLIGHT_TIMEOUT": "0.1"},
            expected_connection=("127.0.0.1", 23456, 0.1),
        )
        self.assertNotEqual(0, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("недоступен", result.stderr)

    def test_unreachable_external_proxy_never_runs_child(self):
        self.write_config(23456)
        marker = self.config_home / "external-child-ran"
        helper = self.config_home / "external-child"
        helper.write_text(
            "#!/usr/bin/env bash\nprintf ran > \"$1\"\n", encoding="ascii"
        )
        helper.chmod(0o755)

        result = self.run_cli_with_socket_setup(
            ("run", "--", str(helper), str(marker)),
            REFUSED_SOCKET_SETUP,
            {"PROXY_PREFLIGHT_TIMEOUT": "0.1"},
            expected_connection=("127.0.0.1", 23456, 0.1),
        )

        self.assertNotEqual(0, result.returncode)
        self.assertIn("недоступен", result.stderr)
        self.assertFalse(marker.exists())

    def test_xray_type_fails_closed_for_all_ticket_commands(self):
        self.write_config(12334, proxy_type="xray")
        marker = self.config_home / "child-ran"
        helper = self.config_home / "child"
        helper.write_text(
            "#!/usr/bin/env bash\nprintf ran > \"$1\"\n", encoding="ascii"
        )
        helper.chmod(0o755)
        commands = [
            ("status",),
            ("env",),
            ("run", "--", str(helper), str(marker)),
            ("start",),
        ]
        for command in commands:
            with self.subTest(command=command):
                result = self.run_cli(*command)
                self.assertNotEqual(0, result.returncode)
                self.assertEqual("", result.stdout)
                self.assertIn("Ошибка:", result.stderr)
                self.assertFalse(marker.exists())

    def test_external_start_fails_without_starting_anything(self):
        self.write_config(12334)
        result = self.run_cli("start")
        self.assertNotEqual(0, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("только для type=xray", result.stderr)

    def test_run_preserves_arguments_exit_code_and_child_only_environment(self):
        self.write_config(23456)
        code = (
            "import json, os, sys; "
            "print(json.dumps({'args': sys.argv[1:], "
            "'http': os.environ['HTTP_PROXY'], "
            "'lower': os.environ['http_proxy']})); sys.exit(42)"
        )
        original_http = os.environ.get("HTTP_PROXY")
        result = self.run_cli_with_socket_setup(
            ("run", "--", sys.executable, "-c", code, "a b", "--", "--flag"),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("127.0.0.1", 23456, 2.0),
        )

        self.assertEqual(42, result.returncode, result.stderr)
        child = json.loads(result.stdout)
        self.assertEqual(["a b", "--", "--flag"], child["args"])
        expected_url = "http://127.0.0.1:23456"
        self.assertEqual(expected_url, child["http"])
        self.assertEqual(expected_url, child["lower"])
        self.assertEqual(original_http, os.environ.get("HTTP_PROXY"))

    def test_missing_run_command_returns_127(self):
        self.write_config(12334)
        result = self.run_cli("run", "--", "definitely-no-such-command-xyz")
        self.assertEqual(127, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("не найдена в PATH", result.stderr)

    def test_run_maps_child_signal_to_shell_exit_code(self):
        self.write_config(23456)
        result = self.run_cli_with_socket_setup(
            (
                "run",
                "--",
                sys.executable,
                "-c",
                "import os, signal; os.kill(os.getpid(), signal.SIGTERM)",
            ),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("127.0.0.1", 23456, 2.0),
        )
        self.assertEqual(143, result.returncode, result.stderr)

    def test_run_requires_explicit_separator_and_command(self):
        self.write_config(12334)
        for command in (("run",), ("run", "true"), ("run", "--")):
            with self.subTest(command=command):
                result = self.run_cli(*command)
                self.assertNotEqual(0, result.returncode)
                self.assertIn("run -- COMMAND", result.stderr)

    def test_invalid_preflight_timeout_is_rejected_clearly(self):
        self.write_config(12334)
        result = self.run_cli(
            "status", environment={"PROXY_PREFLIGHT_TIMEOUT": "not-a-number"}
        )
        self.assertNotEqual(0, result.returncode)
        self.assertIn("положительным числом", result.stderr)

    def test_ini_rejects_structure_key_and_value_grammar_errors(self):
        valid = (
            "[proxy]\n"
            "type = external\n"
            "name = Test\n"
            "address = 127.0.0.1\n"
            "port = 12334\n"
            "no_proxy_extra =\n"
        )
        invalid_configs = {
            "unknown section": valid + "[other]\nkey = value\n",
            "unknown key": valid + "extra = value\n",
            "duplicate key": valid + "port = 12335\n",
            "missing key": valid.replace("name = Test\n", ""),
            "empty name": valid.replace("name = Test", "name =   "),
            "empty address": valid.replace("address = 127.0.0.1", "address =   "),
            "port zero": valid.replace("port = 12334", "port = 0"),
            "port high": valid.replace("port = 12334", "port = 65536"),
            "port syntax": valid.replace("port = 12334", "port = +80"),
            "empty no-proxy item": valid.replace(
                "no_proxy_extra =", "no_proxy_extra = one.example,,two.example"
            ),
        }
        for label, content in invalid_configs.items():
            with self.subTest(label=label):
                self.write_raw_config(content)
                result = self.run_cli("start")
                self.assertNotEqual(0, result.returncode)
                self.assertEqual("", result.stdout)
                self.assertIn("Ошибка:", result.stderr)

    def test_address_accepts_ip_literals_and_simple_ascii_names(self):
        valid_addresses = (
            "127.0.0.1",
            "2001:db8::1",
            "proxy-1.example",
            ".leading.example",
            "trailing.example.",
            "bad..example",
            "-bad.example",
            "bad-.example",
        )
        for address in valid_addresses:
            with self.subTest(address=address):
                self.write_config(12334, address=address)
                result = self.run_cli("start")
                self.assertIn("только для type=xray", result.stderr)

        invalid_addresses = (
            "http://proxy.example",
            "[::1]",
            "has space",
            "proxy/path",
            "user@proxy",
            "прокси.example",
            "fe80::1%eth0",
        )
        for address in invalid_addresses:
            with self.subTest(address=address):
                self.write_config(12334, address=address)
                result = self.run_cli("start")
                self.assertNotEqual(0, result.returncode)
                self.assertNotIn("только для type=xray", result.stderr)
                self.assertIn("address", result.stderr)


class XrayInstallTest(unittest.TestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary.name)
        self.home = self.root / "home"
        self.state_home = self.root / "state"
        self.home.mkdir()
        self.environment = {
            "HOME": str(self.home),
            "XDG_STATE_HOME": str(self.state_home),
            "PATH": str(self.home / ".local/bin"),
        }
        self.module = runpy.run_path(str(CLI))
        self.destination = self.home / ".local/bin/xray"
        self.archive = self.root / "fixture.zip"
        self.write_archive("installed")

    def tearDown(self):
        self.temporary.cleanup()

    @staticmethod
    def fake_xray(label="fixture", version="26.3.27"):
        return (
            "#!/bin/sh\n"
            "if [ \"$1\" = version ]; then\n"
            "  printf '%s\\n' 'Xray {} ({})'\n"
            "  exit 0\n"
            "fi\n"
            "exit 64\n".format(version, label)
        ).encode("ascii")

    def write_archive(self, label="fixture", version="26.3.27", member="xray"):
        with zipfile.ZipFile(str(self.archive), "w") as archive:
            archive.writestr(member, self.fake_xray(label, version))

    @staticmethod
    def official_digest(archive_hash):
        return (
            "MD5= {}\n"
            "SHA1= {}\n"
            "SHA2-256= {}\n"
            "SHA2-512= {}\n".format(
                "1" * 32,
                "2" * 40,
                archive_hash,
                "3" * 128,
            )
        )

    def downloader(self, bad_hash=False, digest_text=None, fail_after_archive=False):
        calls = []

        def download(url, destination, _max_bytes):
            calls.append(url)
            if url.endswith(".dgst"):
                if fail_after_archive:
                    raise self.module["ProxyError"]("инъецированная ошибка загрузки")
                archive_hash = hashlib.sha256(self.archive.read_bytes()).hexdigest()
                if bad_hash:
                    archive_hash = "0" * 64
                text = digest_text
                if text is None:
                    text = self.official_digest(archive_hash)
                destination.write_text(text, encoding="ascii")
            else:
                destination.write_bytes(self.archive.read_bytes())

        return download, calls

    def install(self, replace=False, downloader=None, machine="x86_64", path=None):
        if downloader is None:
            downloader, _calls = self.downloader()
        environment = dict(self.environment)
        if path is not None:
            environment["PATH"] = path
        globals_ = self.module["command_install"].__globals__
        with mock.patch.dict(os.environ, environment, clear=False):
            with mock.patch.object(globals_["platform"], "machine", return_value=machine):
                return self.module["command_install"](replace, downloader)

    def write_destination(self, content=None, executable=True):
        self.destination.parent.mkdir(parents=True, exist_ok=True)
        self.destination.write_bytes(content or self.fake_xray("existing"))
        self.destination.chmod(0o755 if executable else 0o644)

    def tree_snapshot(self):
        snapshot = {}
        for path in self.root.rglob("*"):
            relative = path.relative_to(self.root).as_posix()
            info = path.lstat()
            mode = info.st_mode & 0o777
            if path.is_symlink():
                snapshot[relative] = ("symlink", mode, os.readlink(str(path)))
            elif path.is_dir():
                snapshot[relative] = ("directory", mode)
            elif path.is_file():
                digest = hashlib.sha256(path.read_bytes()).hexdigest()
                snapshot[relative] = ("file", mode, digest)
            else:
                snapshot[relative] = ("other", mode)
        return snapshot

    def tree_snapshot_without_install_lock(self):
        snapshot = self.tree_snapshot()
        for path in ("state/proxy-client/lock", "state/proxy-client", "state"):
            snapshot.pop(path, None)
        return snapshot

    def assert_no_staging(self):
        if self.destination.parent.exists():
            self.assertEqual(
                [], list(self.destination.parent.glob(".xray-install-*"))
            )

    def assert_install_lock_released(self):
        lock_path = self.state_home / "proxy-client/lock"
        with lock_path.open("r+", encoding="ascii") as lock:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)

    def test_architecture_mapping_is_pure_and_rejects_unknown(self):
        asset_for = self.module["xray_asset_for_architecture"]
        for machine in ("x86_64", "amd64"):
            self.assertEqual("Xray-linux-64.zip", asset_for(machine))
        for machine in ("aarch64", "arm64"):
            self.assertEqual("Xray-linux-arm64-v8a.zip", asset_for(machine))
        with self.assertRaisesRegex(self.module["ProxyError"], "не поддерживается"):
            asset_for("riscv64")

    def test_valid_digest_installs_verified_executable_and_only_expected_files(self):
        downloader, calls = self.downloader()
        self.assertEqual(0, self.install(downloader=downloader))

        self.assertEqual(0o755, self.destination.stat().st_mode & 0o777)
        self.assertIn(b"installed", self.destination.read_bytes())
        self.assertEqual(2, len(calls))
        self.assertTrue(calls[0].endswith("/Xray-linux-64.zip"))
        self.assertEqual(calls[0] + ".dgst", calls[1])
        self.assertEqual(
            {
                "fixture.zip",
                "home",
                "home/.local",
                "home/.local/bin",
                "home/.local/bin/xray",
                "state",
                "state/proxy-client",
                "state/proxy-client/lock",
            },
            set(self.tree_snapshot()),
        )
        self.assert_no_staging()

    def test_platform_machine_selects_arm64_asset(self):
        downloader, calls = self.downloader()
        self.assertEqual(
            0, self.install(downloader=downloader, machine="aarch64")
        )
        self.assertTrue(calls[0].endswith("/Xray-linux-arm64-v8a.zip"))
        self.assertEqual(calls[0] + ".dgst", calls[1])

    def test_candidate_is_private_while_zip_bytes_are_written(self):
        os_module = self.module["extract_xray_candidate"].__globals__["os"]
        real_fdopen = os_module.fdopen
        observed_modes = []

        def checked_fdopen(descriptor, *arguments, **keywords):
            descriptor_path = os.readlink("/proc/self/fd/{}".format(descriptor))
            if Path(descriptor_path).name == "xray":
                observed_modes.append(os.fstat(descriptor).st_mode & 0o777)
            return real_fdopen(descriptor, *arguments, **keywords)

        with mock.patch.object(
            os_module, "fdopen", side_effect=checked_fdopen
        ):
            self.assertEqual(0, self.install())

        self.assertEqual([0o600], observed_modes)
        self.assertEqual(0o755, self.destination.stat().st_mode & 0o777)

    def test_bad_hash_and_digest_format_preserve_destination_before_extraction(self):
        valid_hash = "a" * 64
        cases = (
            self.downloader(bad_hash=True)[0],
            self.downloader(digest_text="SHA256= " + valid_hash + "\n")[0],
            self.downloader(
                digest_text=self.official_digest(valid_hash).replace(
                    "SHA2-256= " + valid_hash + "\n", ""
                )
            )[0],
            self.downloader(
                digest_text=self.official_digest(valid_hash)
                + "SHA2-256= "
                + valid_hash
                + "\n"
            )[0],
            self.downloader(
                digest_text=self.official_digest(valid_hash) + "garbage\n"
            )[0],
            self.downloader(
                digest_text=self.official_digest(valid_hash).replace(
                    valid_hash, "a" * 63
                )
            )[0],
            self.downloader(
                digest_text=self.official_digest(valid_hash).replace(
                    valid_hash, "a" * 65
                )
            )[0],
        )
        for downloader in cases:
            with self.subTest(downloader=downloader):
                sentinel = b"destination sentinel"
                self.write_destination(sentinel)
                before = self.tree_snapshot_without_install_lock()
                with self.assertRaises(self.module["ProxyError"]):
                    self.install(replace=True, downloader=downloader)
                self.assertEqual(sentinel, self.destination.read_bytes())
                self.assert_no_staging()
                self.assertEqual(before, self.tree_snapshot_without_install_lock())

    def test_bad_or_truncated_archive_is_rejected_before_zip_open(self):
        full_archive = self.archive.read_bytes()

        def truncated_downloader(url, destination, _max_bytes):
            if url.endswith(".dgst"):
                digest = hashlib.sha256(full_archive).hexdigest()
                destination.write_text(self.official_digest(digest), encoding="ascii")
            else:
                destination.write_bytes(full_archive[: len(full_archive) // 2])

        zip_file = self.module["extract_xray_candidate"].__globals__["zipfile"]
        with mock.patch.object(
            zip_file, "ZipFile", side_effect=AssertionError("ZIP открыт до SHA-256")
        ) as opened:
            with self.assertRaisesRegex(self.module["ProxyError"], "SHA-256"):
                self.install(downloader=truncated_downloader)
        opened.assert_not_called()
        self.assertFalse(self.destination.exists())
        self.assert_no_staging()

    def test_real_downloader_installs_complete_declared_responses(self):
        archive = self.archive.read_bytes()
        digest = self.official_digest(hashlib.sha256(archive).hexdigest()).encode(
            "ascii"
        )
        responses = [
            FakeUrlResponse(archive, content_length=len(archive)),
            FakeUrlResponse(digest, content_length=len(digest)),
        ]
        urllib_request = self.module["download_file"].__globals__["urllib"].request

        with mock.patch.object(urllib_request, "urlopen", side_effect=responses):
            self.assertEqual(
                0, self.install(downloader=self.module["download_file"])
            )

    def test_real_downloader_refuses_length_errors_and_releases_install(self):
        archive = self.archive.read_bytes()
        cases = (
            (
                "early EOF",
                FakeUrlResponse(archive, content_length=len(archive) + 1),
            ),
            (
                "extra bytes",
                FakeUrlResponse(archive, content_length=len(archive) - 1),
            ),
            (
                "declared oversized",
                FakeUrlResponse(
                    b"",
                    content_length=self.module["MAX_ARCHIVE_BYTES"] + 1,
                ),
            ),
            ("url error", urllib.error.URLError("offline")),
            (
                "http error",
                FakeUrlResponse(read_error=http.client.HTTPException("broken HTTP")),
            ),
            (
                "incomplete read",
                FakeUrlResponse(
                    read_error=http.client.IncompleteRead(b"partial", 10)
                ),
            ),
        )
        urllib_request = self.module["download_file"].__globals__["urllib"].request

        for label, outcome in cases:
            with self.subTest(label=label):
                sentinel = b"destination sentinel"
                self.write_destination(sentinel)
                before = self.tree_snapshot_without_install_lock()
                with mock.patch.object(urllib_request, "urlopen", side_effect=[outcome]):
                    with self.assertRaises(self.module["ProxyError"]):
                        self.install(
                            replace=True, downloader=self.module["download_file"]
                        )
                self.assertEqual(sentinel, self.destination.read_bytes())
                self.assert_no_staging()
                self.assert_install_lock_released()
                self.assertEqual(before, self.tree_snapshot_without_install_lock())

    def test_real_downloader_counts_limit_plus_one_without_content_length(self):
        urllib_request = self.module["download_file"].__globals__["urllib"].request
        globals_ = self.module["command_install"].__globals__
        sentinel = b"destination sentinel"
        self.write_destination(sentinel)

        with mock.patch.dict(globals_, {"MAX_ARCHIVE_BYTES": 8}):
            response = FakeUrlResponse(b"123456789")
            with mock.patch.object(urllib_request, "urlopen", return_value=response):
                with self.assertRaisesRegex(self.module["ProxyError"], "размер"):
                    self.install(
                        replace=True, downloader=self.module["download_file"]
                    )

        self.assertEqual(sentinel, self.destination.read_bytes())
        self.assert_no_staging()
        self.assert_install_lock_released()

    def test_uppercase_digest_is_rejected_before_zip_open(self):
        digest = hashlib.sha256(self.archive.read_bytes()).hexdigest().upper()
        downloader, _calls = self.downloader(
            digest_text=self.official_digest(digest)
        )
        zip_file = self.module["extract_xray_candidate"].__globals__["zipfile"]
        with mock.patch.object(
            zip_file, "ZipFile", side_effect=AssertionError("ZIP открыт до digest")
        ) as opened:
            with self.assertRaisesRegex(self.module["ProxyError"], "SHA-256"):
                self.install(downloader=downloader)
        opened.assert_not_called()
        self.assertFalse(self.destination.exists())
        self.assert_no_staging()

    def test_existing_verified_install_is_noop_without_downloader(self):
        self.write_destination()
        downloader = mock.Mock(side_effect=AssertionError("network used"))
        self.assertEqual(0, self.install(downloader=downloader))
        downloader.assert_not_called()

    def test_existing_version_gate_keeps_timeout_and_avoids_downloader(self):
        self.write_destination()
        downloader = mock.Mock(side_effect=AssertionError("network used"))
        subprocess_module = self.module["xray_version"].__globals__["subprocess"]
        timeout = subprocess.TimeoutExpired([str(self.destination), "version"], 5)

        with mock.patch.object(subprocess_module, "run", side_effect=timeout) as run:
            with self.assertRaisesRegex(self.module["ProxyError"], "--replace"):
                self.install(downloader=downloader)

        downloader.assert_not_called()
        self.assertEqual(5, run.call_args.kwargs["timeout"])
        self.assertEqual(self.fake_xray("existing"), self.destination.read_bytes())

    def test_replace_reinstalls_verified_existing_file(self):
        self.write_destination(self.fake_xray("old"))
        downloader, calls = self.downloader()
        self.assertEqual(0, self.install(replace=True, downloader=downloader))
        self.assertIn(b"installed", self.destination.read_bytes())
        self.assertEqual(2, len(calls))

    def test_destination_table_for_symlink_directory_and_bad_files(self):
        self.destination.parent.mkdir(parents=True)
        target = self.root / "target"
        target.write_bytes(b"target sentinel")
        self.destination.symlink_to(target)
        with self.assertRaisesRegex(self.module["ProxyError"], "--replace"):
            self.install()
        self.assertTrue(self.destination.is_symlink())
        self.assertEqual(0, self.install(replace=True))
        self.assertFalse(self.destination.is_symlink())
        self.assertEqual(b"target sentinel", target.read_bytes())

        self.destination.unlink()
        self.destination.mkdir()
        for replace in (False, True):
            with self.subTest(directory=True, replace=replace):
                with self.assertRaisesRegex(self.module["ProxyError"], "каталог"):
                    self.install(replace=replace)
        self.destination.rmdir()

        for content, executable in ((b"broken", True), (self.fake_xray(), False)):
            with self.subTest(content=content, executable=executable):
                self.write_destination(content, executable)
                with self.assertRaisesRegex(self.module["ProxyError"], "--replace"):
                    self.install()
                self.assertEqual(0, self.install(replace=True))
                self.assertIn(b"installed", self.destination.read_bytes())
                self.destination.unlink()

    def test_parent_symlink_cannot_redirect_install_outside_home(self):
        outside = self.root / "outside"
        outside.mkdir()
        (self.home / ".local").symlink_to(outside, target_is_directory=True)
        downloader = mock.Mock()

        with self.assertRaisesRegex(self.module["ProxyError"], "домашнего каталога"):
            self.install(downloader=downloader)

        downloader.assert_not_called()
        self.assertEqual([], list(outside.iterdir()))

    def test_managed_state_guard_only_allows_absent_or_stale(self):
        guard = self.module["require_install_safe_managed_state"]
        globals_ = guard.__globals__
        present = mock.Mock()
        present.lstat.return_value = mock.Mock()
        with mock.patch.dict(
            globals_,
            {
                "state_path": mock.Mock(return_value=present),
                "read_managed_state": mock.Mock(return_value={"state": "fixture"}),
            },
        ):
            for outcome in ("healthy", "orphan", "unverifiable"):
                with self.subTest(outcome=outcome):
                    globals_["managed_state_outcome"] = mock.Mock(return_value=outcome)
                    with self.assertRaisesRegex(self.module["ProxyError"], "stop"):
                        guard()
            globals_["managed_state_outcome"] = mock.Mock(return_value="stale")
            guard()

        absent = mock.Mock()
        absent.lstat.side_effect = FileNotFoundError()
        with mock.patch.dict(
            globals_, {"state_path": mock.Mock(return_value=absent)}
        ):
            guard()

    def test_command_classifies_managed_state_before_even_noop_destination(self):
        self.write_destination()
        downloader = mock.Mock(side_effect=AssertionError("network used"))
        guard = mock.Mock(
            side_effect=self.module["ProxyError"](
                "managed Xray работает; сначала выполните proxy-client stop"
            )
        )
        globals_ = self.module["command_install"].__globals__

        with mock.patch.dict(
            globals_, {"require_install_safe_managed_state": guard}
        ):
            with self.assertRaisesRegex(self.module["ProxyError"], "stop"):
                self.install(downloader=downloader)

        guard.assert_called_once_with()
        downloader.assert_not_called()
        self.assertEqual(self.fake_xray("existing"), self.destination.read_bytes())

    def test_download_extraction_and_version_errors_are_atomic_and_clean_staging(self):
        failures = []
        failures.append(self.downloader(fail_after_archive=True)[0])
        missing_member = self.root / "missing-member.zip"
        with zipfile.ZipFile(str(missing_member), "w") as archive:
            archive.writestr("not-xray", b"data")

        def missing_member_downloader(url, destination, _max_bytes):
            source = missing_member if not url.endswith(".dgst") else None
            if source is not None:
                destination.write_bytes(source.read_bytes())
            else:
                digest = hashlib.sha256(missing_member.read_bytes()).hexdigest()
                destination.write_text(self.official_digest(digest), encoding="ascii")

        failures.append(missing_member_downloader)
        self.write_archive("wrong", version="26.3.26")
        failures.append(self.downloader()[0])

        for downloader in failures:
            with self.subTest(downloader=downloader):
                sentinel = b"destination sentinel"
                self.write_destination(sentinel)
                before = self.tree_snapshot_without_install_lock()
                with self.assertRaises(self.module["ProxyError"]):
                    self.install(replace=True, downloader=downloader)
                self.assertEqual(sentinel, self.destination.read_bytes())
                self.assert_no_staging()
                self.assertEqual(before, self.tree_snapshot_without_install_lock())

    def test_corrupt_zip_with_matching_digest_is_atomic_and_cleaned(self):
        corrupt = b"not a ZIP archive"

        def corrupt_downloader(url, destination, _max_bytes):
            if url.endswith(".dgst"):
                digest = hashlib.sha256(corrupt).hexdigest()
                destination.write_text(self.official_digest(digest), encoding="ascii")
            else:
                destination.write_bytes(corrupt)

        sentinel = b"destination sentinel"
        self.write_destination(sentinel)
        before = self.tree_snapshot_without_install_lock()
        with self.assertRaisesRegex(self.module["ProxyError"], "извлечь"):
            self.install(replace=True, downloader=corrupt_downloader)
        self.assertEqual(sentinel, self.destination.read_bytes())
        self.assert_no_staging()
        self.assertEqual(before, self.tree_snapshot_without_install_lock())

    def test_injected_oversized_archive_and_digest_are_bounded(self):
        globals_ = self.module["command_install"].__globals__
        cases = (
            ("MAX_ARCHIVE_BYTES", len(self.archive.read_bytes()) - 1, self.downloader()[0]),
            (
                "MAX_DIGEST_BYTES",
                8,
                self.downloader(digest_text=self.official_digest("a" * 64))[0],
            ),
        )
        for constant, limit, downloader in cases:
            with self.subTest(constant=constant):
                sentinel = b"destination sentinel"
                self.write_destination(sentinel)
                with mock.patch.dict(globals_, {constant: limit}):
                    with self.assertRaisesRegex(self.module["ProxyError"], "размер"):
                        self.install(replace=True, downloader=downloader)
                self.assertEqual(sentinel, self.destination.read_bytes())
                self.assert_no_staging()

    def test_zip_member_declared_over_limit_is_rejected_before_extraction(self):
        globals_ = self.module["command_install"].__globals__
        binary_size = len(self.fake_xray())
        sentinel = b"destination sentinel"
        self.write_destination(sentinel)

        with mock.patch.dict(globals_, {"MAX_XRAY_BYTES": binary_size - 1}):
            with self.assertRaisesRegex(self.module["ProxyError"], "размер"):
                self.install(replace=True)

        self.assertEqual(sentinel, self.destination.read_bytes())
        self.assert_no_staging()

    def test_streamed_zip_output_cannot_exceed_limit_even_if_metadata_lies(self):
        class FakeInfo:
            filename = "xray"
            file_size = 8

            @staticmethod
            def is_dir():
                return False

        class FakeArchive:
            def __enter__(self):
                return self

            def __exit__(self, _type, _value, _traceback):
                return False

            @staticmethod
            def infolist():
                return [FakeInfo()]

            @staticmethod
            def open(_member):
                return FakeUrlResponse(b"123456789")

        globals_ = self.module["command_install"].__globals__
        zip_file = globals_["zipfile"]
        sentinel = b"destination sentinel"
        self.write_destination(sentinel)
        with mock.patch.dict(globals_, {"MAX_XRAY_BYTES": 8}):
            with mock.patch.object(zip_file, "ZipFile", return_value=FakeArchive()):
                with self.assertRaisesRegex(self.module["ProxyError"], "размер"):
                    self.install(replace=True)
        self.assertEqual(sentinel, self.destination.read_bytes())
        self.assert_no_staging()

    def test_interruption_preserves_destination_cleans_staging_and_releases_lock(self):
        sentinel = b"destination sentinel"
        self.write_destination(sentinel)
        before = self.tree_snapshot_without_install_lock()

        def interrupted_downloader(_url, destination, _max_bytes):
            destination.write_bytes(b"partial")
            raise KeyboardInterrupt()

        with self.assertRaises(KeyboardInterrupt):
            self.install(replace=True, downloader=interrupted_downloader)

        self.assertEqual(sentinel, self.destination.read_bytes())
        self.assert_no_staging()
        lock_path = self.state_home / "proxy-client/lock"
        with lock_path.open("r+", encoding="ascii") as lock:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        self.assertEqual(before, self.tree_snapshot_without_install_lock())

    def test_start_lock_is_held_during_download_and_through_replace(self):
        downloader, _calls = self.downloader()
        lock_path = self.state_home / "proxy-client/lock"

        def assert_lock_held():
            with lock_path.open("r+", encoding="ascii") as competing:
                with self.assertRaises(BlockingIOError):
                    fcntl.flock(competing, fcntl.LOCK_EX | fcntl.LOCK_NB)

        def checked_downloader(url, destination, max_bytes):
            assert_lock_held()
            downloader(url, destination, max_bytes)

        os_module = self.module["command_install"].__globals__["os"]
        real_replace = os_module.replace

        def checked_replace(source, destination):
            real_replace(source, destination)
            assert_lock_held()

        with mock.patch.object(os_module, "replace", side_effect=checked_replace):
            self.assertEqual(0, self.install(downloader=checked_downloader))

        with lock_path.open("r+", encoding="ascii") as released:
            fcntl.flock(released, fcntl.LOCK_EX | fcntl.LOCK_NB)

    def test_unsupported_architecture_refuses_before_downloading(self):
        downloader = mock.Mock()
        with self.assertRaisesRegex(self.module["ProxyError"], "не поддерживается"):
            self.install(downloader=downloader, machine="riscv64")
        downloader.assert_not_called()
        self.assertFalse(self.destination.exists())

    def test_path_missing_or_competing_returns_error_without_rollback(self):
        cases = (("empty", False), ("competitor", True))
        for directory_name, competing in cases:
            with self.subTest(directory_name=directory_name):
                path_dir = self.root / directory_name
                path_dir.mkdir()
                if competing:
                    competitor = path_dir / "xray"
                    competitor.write_bytes(self.fake_xray("competitor"))
                    competitor.chmod(0o755)
                    path = str(path_dir) + os.pathsep + str(self.destination.parent)
                else:
                    path = str(path_dir)
                with self.assertRaisesRegex(self.module["ProxyError"], "PATH"):
                    self.install(path=path)
                self.assertTrue(self.destination.is_file())
                self.destination.unlink()

    def test_busy_install_hint_uses_exception_type_not_message_text(self):
        globals_ = self.module["acquire_install_lock"].__globals__
        busy_error = self.module["StartLockBusyError"]("completely changed text")
        with mock.patch.dict(
            globals_, {"acquire_start_lock": mock.Mock(side_effect=busy_error)}
        ):
            with self.assertRaisesRegex(self.module["ProxyError"], "stop"):
                self.module["acquire_install_lock"]()

    def test_path_comparison_handles_relative_entry_and_internal_bin_symlink(self):
        local = self.home / ".local"
        actual_bin = self.home / "private-bin"
        local.mkdir()
        actual_bin.mkdir()
        (local / "bin").symlink_to(actual_bin, target_is_directory=True)

        self.assertEqual(0, self.install())
        self.assertTrue((actual_bin / "xray").is_file())

        previous_cwd = Path.cwd()
        try:
            os.chdir(str(actual_bin))
            self.assertEqual(0, self.install(path="."))
        finally:
            os.chdir(str(previous_cwd))

    def test_install_dispatches_before_machine_config_loading(self):
        self.write_destination()
        config = self.root / "config/proxy-client"
        config.mkdir(parents=True)
        (config / "config").write_text("not an ini\n", encoding="ascii")
        environment = os.environ.copy()
        environment.update(self.environment)
        environment["PATH"] = self.environment["PATH"] + os.pathsep + os.environ["PATH"]
        environment["XDG_CONFIG_HOME"] = str(self.root / "config")
        result = subprocess.run(
            [str(CLI), "install"],
            text=True,
            capture_output=True,
            env=environment,
            check=False,
        )
        self.assertEqual(0, result.returncode, result.stderr)
        self.assertIn("уже установлен", result.stdout)


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