From fc95b79092e7b37baa3da43555811eaf136d3b99 Mon Sep 17 00:00:00 2001 From: chuan Date: Thu, 28 May 2026 14:19:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8F=90=E4=BE=9B=E5=91=BD=E4=BB=A4?= =?UTF-8?q?=E8=A1=8C=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 31 +++++- design.md | 0 docs/config/assets.md | 6 +- pyxray/cli.py | 103 +++++++++++++++++++- pyxray/libs/app_data.py | 149 +++++++++++++++++++++++++++++ pyxray/libs/xray_asset_settings.py | 6 +- pyxray/libs/xray_assets.py | 75 +++++++++++++-- pyxray/web/dashboard.py | 4 +- pyxray/web/server.py | 20 ++-- pyxray/web/xray_assets.py | 5 +- tests/libs/test_xray_assets.py | 65 +++++++++++-- tests/libs/test_xray_runtime.py | 22 ++++- tests/test_cli.py | 138 ++++++++++++++++++++++++++ tests/web/test_xray_assets_web.py | 137 ++++++++++++++++---------- uv.lock | 2 +- 15 files changed, 661 insertions(+), 102 deletions(-) create mode 100644 design.md create mode 100644 pyxray/libs/app_data.py create mode 100644 tests/test_cli.py diff --git a/README.md b/README.md index 918276a..1e2e501 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ flowchart LR | `data/settings.toml` | 保存配置页设置。 | | `data/download.toml` | 保存下载页设置。 | | `data/config.json` | 生成给 Xray 使用的配置。 | -| `data/xray/` | 保存 `xray`、`geoip.dat`、`geosite.dat`。 | +| `data/xray/` | 保存 `xray` / `xray.exe`、`geoip.dat`、`geosite.dat`。 | | `data/transparent/` | 保存透明代理脚本、nftables 配置和 `tinytun.yaml`。 | | `data/xray.log` | 保存 Xray 输出和 pyxray 运行日志。 | @@ -82,6 +82,35 @@ http://:8080 7. 点击“启动 Xray”。 8. 在“日志”页确认 Xray 和透明代理脚本执行结果。 +## CLI + +启动 Web 控制台： + +```bash +pyxray web --host 127.0.0.1 --port 3309 --xray-dir data/xray +``` + +查看配置文件： + +```bash +pyxray configs --download +pyxray configs --settings +``` + +清理配置/下载资源： + +```bash +pyxray clear --download +pyxray clear --all +``` + +直接下载或补齐 Xray 资源： + +```bash +pyxray download --target all --directory data/xray --force +pyxray download --target geoip --geoip-url https://example.invalid/geoip.dat +``` + ## 透明代理建议 | 场景 | 建议 | diff --git a/design.md b/design.md new file mode 100644 index 0000000..e69de29 diff --git a/docs/config/assets.md b/docs/config/assets.md index 0700b06..df8883d 100644 --- a/docs/config/assets.md +++ b/docs/config/assets.md @@ -5,7 +5,7 @@ | 设置 | UI | 默认值 | 可选值 | 作用 | 什么时候修改 | | --- | --- | --- | --- | --- | --- | | `directory` | 下载页 | `data/xray`；Docker 中通常为 `/config/xray` | 路径 | Xray 资源保存目录。 | Docker 部署保持 `/config/xray`；本机运行可用默认值。 | -| `version` | 下载页 | `v26.5.9` | Xray release tag | 官方 release 版本。 | 需要固定或升级 Xray 版本时修改。 | +| `version` | 下载页 | 首次打开时优先使用最新 release；获取失败回退 `v26.5.9` | Xray release tag | 官方 release 版本。 | 需要固定或升级 Xray 版本时修改。 | | `archive_url` | 下载页 | `""` | URL | 自定义 Xray release zip 地址；为空时用官方地址。 | 官方下载慢或使用镜像时修改。 | | `geoip_url` | 下载页 | `""` | URL | 自定义 `geoip.dat` 下载地址。 | 需要替换 geoip 数据源时修改。 | | `geosite_url` | 下载页 | `""` | URL | 自定义 `geosite.dat` 下载地址。 | 需要替换 geosite 数据源时修改。 | @@ -17,7 +17,7 @@ | 文件 | 作用 | | --- | --- | -| `xray` | Xray 可执行文件。 | +| `xray` / `xray.exe` | Xray 可执行文件；Windows 下为 `xray.exe`，其它平台为 `xray`。 | | `geoip.dat` | IP 地理库，用于 `geoip:*` 规则。 | | `geosite.dat` | 域名分类库，用于 `geosite:*` 规则。 | @@ -26,7 +26,9 @@ | 条件 | 行为 | | --- | --- | | `target = all` | 确保三个必需文件都存在。 | +| 首次无 `download.toml` | 尝试在 5 秒内获取 Xray-core 最新 release tag；失败则使用内置回退版本。 | | `force = false` 且文件存在 | 跳过已有文件。 | | `force = true` | 覆盖目标文件。 | +| `archive_url` 为空 | 按当前平台选择官方 release zip；Windows x64 使用 `Xray-windows-64.zip`，Linux x64 使用 `Xray-linux-64.zip`。 | | `geoip_url` / `geosite_url` 非空 | 对应 dat 文件使用自定义 URL，优先于 release zip 内置版本。 | | Docker 部署 | 资源仍由 Web 下载页处理，不在 Dockerfile 中下载。 | diff --git a/pyxray/cli.py b/pyxray/cli.py index 1df19b5..07d82cf 100644 --- a/pyxray/cli.py +++ b/pyxray/cli.py @@ -1,30 +1,123 @@ from __future__ import annotations import argparse +from pathlib import Path +from pyxray.libs.app_data import ( + CONFIG_FILES, + clear_all_data, + clear_download_data, + download_xray_assets, + read_config_file, + resolve_app_data_paths, +) +from pyxray.libs.xray_assets import ASSET_TARGETS from pyxray.web.server import run_web +DEFAULT_HOST = "0.0.0.0" +DEFAULT_PORT = 8000 +DEFAULT_XRAY_DIR = "data/xray" + + def main(argv: list[str] | None = None) -> None: """pyxray 命令行入口。""" parser = argparse.ArgumentParser(prog="pyxray") subparsers = parser.add_subparsers(dest="command") _add_web_parser(subparsers) + _add_clear_parser(subparsers) + _add_configs_parser(subparsers) + _add_download_parser(subparsers) args = parser.parse_args(argv) - if args.command in (None, "web"): - run_web(args.host, args.port, args.xray_dir) + if args.command is None: + run_web(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_XRAY_DIR) + return + args.func(args) def _add_web_parser(subparsers: argparse._SubParsersAction) -> None: """注册 Web 服务启动参数。""" parser = subparsers.add_parser("web", help="启动 Web 控制台") - parser.add_argument("--host", default="0.0.0.0", help="监听地址，默认 0.0.0.0") - parser.add_argument("--port", default=8000, type=int, help="监听端口，默认 8000") - parser.add_argument("--xray-dir", default="data/xray", help="Xray 资源目录，默认 data/xray") + parser.add_argument("--host", default=DEFAULT_HOST, help=f"监听地址，默认 {DEFAULT_HOST}") + parser.add_argument("--port", default=DEFAULT_PORT, type=int, help=f"监听端口，默认 {DEFAULT_PORT}") + parser.add_argument("--xray-dir", default=DEFAULT_XRAY_DIR, help=f"Xray 资源目录，默认 {DEFAULT_XRAY_DIR}") parser.set_defaults(command="web") + parser.set_defaults(func=_run_web_command) + + +def _add_clear_parser(subparsers: argparse._SubParsersAction) -> None: + parser = subparsers.add_parser("clear", help="清除 pyxray 配置/数据文件") + scope = parser.add_mutually_exclusive_group(required=True) + scope.add_argument("--all", action="store_true", help="清除所有 pyxray 配置/数据文件和已知生成产物") + scope.add_argument("--download", action="store_true", help="清除下载设置和已知 Xray 下载产物") + parser.add_argument("--xray-dir", default=DEFAULT_XRAY_DIR, help=f"Xray 资源目录，默认 {DEFAULT_XRAY_DIR}") + parser.set_defaults(command="clear", func=_run_clear_command) + + +def _add_configs_parser(subparsers: argparse._SubParsersAction) -> None: + parser = subparsers.add_parser("configs", help="显示指定 pyxray 配置文件内容") + group = parser.add_mutually_exclusive_group(required=True) + for option in CONFIG_FILES: + group.add_argument(f"--{option.replace('_', '-')}", action="store_const", const=option, dest="config_name") + parser.add_argument("--xray-dir", default=DEFAULT_XRAY_DIR, help=f"Xray 资源目录，默认 {DEFAULT_XRAY_DIR}") + parser.set_defaults(command="configs", func=_run_configs_command) + + +def _add_download_parser(subparsers: argparse._SubParsersAction) -> None: + parser = subparsers.add_parser("download", help="下载或补齐 Xray 运行资源") + parser.add_argument("--target", choices=ASSET_TARGETS, default=None, help="下载目标：all、xray、geoip、geosite") + parser.add_argument("--directory", default=None, help=f"Xray 资源目录，默认读取 download.toml 或 {DEFAULT_XRAY_DIR}") + parser.add_argument("--version", default=None, help="Xray-core release 版本，例如 v26.5.9") + parser.add_argument("--force", action="store_true", default=None, help="覆盖并重新下载已存在文件") + parser.add_argument("--archive-url", default=None, help="自定义 Xray release zip URL") + parser.add_argument("--geoip-url", default=None, help="自定义 geoip.dat URL") + parser.add_argument("--geosite-url", default=None, help="自定义 geosite.dat URL") + parser.add_argument("--proxy-url", default=None, help="下载代理 URL") + parser.set_defaults(command="download", func=_run_download_command) + + +def _run_web_command(args: argparse.Namespace) -> None: + run_web(args.host, args.port, args.xray_dir) + + +def _run_clear_command(args: argparse.Namespace) -> None: + paths = resolve_app_data_paths(args.xray_dir) + result = clear_all_data(paths) if args.all else clear_download_data(paths) + for path in result.removed: + print(f"removed {path}") + if not result.removed: + print("nothing removed") + + +def _run_configs_command(args: argparse.Namespace) -> None: + paths = resolve_app_data_paths(args.xray_dir) + try: + print(read_config_file(paths, args.config_name), end="") + except FileNotFoundError as exc: + raise SystemExit(f"config file not found: {Path(exc.filename)}") from exc + + +def _run_download_command(args: argparse.Namespace) -> None: + directory = args.directory or DEFAULT_XRAY_DIR + paths = resolve_app_data_paths(directory) + overrides = { + "directory": args.directory, + "version": args.version, + "archive_url": args.archive_url, + "geoip_url": args.geoip_url, + "geosite_url": args.geosite_url, + "proxy_url": args.proxy_url, + "target": args.target, + "force": args.force, + } + assets = download_xray_assets(paths.download_settings, default_directory=directory, overrides=overrides) + print(f"directory: {assets.directory}") + print(f"downloaded: {', '.join(assets.downloaded) if assets.downloaded else 'none'}") + print(f"skipped: {', '.join(assets.skipped) if assets.skipped else 'none'}") + print(f"ready: {assets.ready}") if __name__ == "__main__": diff --git a/pyxray/libs/app_data.py b/pyxray/libs/app_data.py new file mode 100644 index 0000000..5be7ec9 --- /dev/null +++ b/pyxray/libs/app_data.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +import shutil +from dataclasses import dataclass +from pathlib import Path + +from pyxray.libs.xray_asset_settings import XrayAssetSettings, XrayAssetSettingsStore +from pyxray.libs.xray_assets import ensure_xray_assets + + +CONFIG_FILES = { + "download": "download.toml", + "nodes": "nodes.toml", + "settings": "settings.toml", + "config": "config.json", + "service_state": "service-state.json", + "log": "xray.log", +} +GENERATED_DIRS = ("transparent",) +DOWNLOAD_ASSET_FILES = ("xray", "xray.exe", "geoip.dat", "geosite.dat") + + +@dataclass(frozen=True, slots=True) +class AppDataPaths: + """Filesystem layout shared by the Web app and CLI.""" + + data_dir: Path + xray_dir: Path + + @property + def download_settings(self) -> Path: + return self.data_dir / CONFIG_FILES["download"] + + @property + def nodes(self) -> Path: + return self.data_dir / CONFIG_FILES["nodes"] + + @property + def settings(self) -> Path: + return self.data_dir / CONFIG_FILES["settings"] + + @property + def generated_config(self) -> Path: + return self.data_dir / CONFIG_FILES["config"] + + @property + def service_state(self) -> Path: + return self.data_dir / CONFIG_FILES["service_state"] + + @property + def log(self) -> Path: + return self.data_dir / CONFIG_FILES["log"] + + @property + def transparent_dir(self) -> Path: + return self.data_dir / "transparent" + + def config_path(self, name: str) -> Path: + try: + return self.data_dir / CONFIG_FILES[name] + except KeyError as exc: + raise ValueError(f"unsupported config name: {name}") from exc + + +@dataclass(frozen=True, slots=True) +class ClearResult: + removed: tuple[Path, ...] + missing: tuple[Path, ...] + + +def resolve_app_data_paths(xray_dir: str | Path = "data/xray", *, data_dir: str | Path | None = None) -> AppDataPaths: + """Resolve the pyxray data directory from the Xray asset directory.""" + + resolved_xray_dir = Path(xray_dir) + resolved_data_dir = Path(data_dir) if data_dir is not None else default_data_dir(resolved_xray_dir) + return AppDataPaths(data_dir=resolved_data_dir, xray_dir=resolved_xray_dir) + + +def default_data_dir(xray_dir: str | Path) -> Path: + """Use data/xray -> data, matching the Web app's default layout.""" + + path = Path(xray_dir) + return path.parent if path.name == "xray" else path + + +def clear_download_data(paths: AppDataPaths) -> ClearResult: + """Remove persisted download settings and known downloaded Xray assets only.""" + + return _remove_known_paths([paths.download_settings, *_download_asset_paths(paths.xray_dir)]) + + +def clear_all_data(paths: AppDataPaths) -> ClearResult: + """Remove pyxray-owned config/data files and generated artifacts.""" + + targets = [ + *(paths.data_dir / filename for filename in CONFIG_FILES.values()), + *(paths.data_dir / name for name in GENERATED_DIRS), + *_download_asset_paths(paths.xray_dir), + ] + return _remove_known_paths(targets) + + +def read_config_file(paths: AppDataPaths, name: str) -> str: + path = paths.config_path(name) + if not path.exists(): + raise FileNotFoundError(path) + return path.read_text(encoding="utf-8") + + +def download_xray_assets(settings_path: str | Path, *, default_directory: str | Path, overrides: dict[str, object]): + """Merge CLI options with download.toml, persist them, and ensure assets exist.""" + + store = XrayAssetSettingsStore(settings_path, default_directory=default_directory) + current = store.load() + values = current.to_dict() + for key, value in overrides.items(): + if value is not None: + values[key] = value + settings = XrayAssetSettings.from_dict(values) + store.save(settings) + return ensure_xray_assets( + settings.directory, + version=settings.version, + archive_url=settings.archive_url or None, + geoip_url=settings.geoip_url or None, + geosite_url=settings.geosite_url or None, + proxy_url=settings.proxy_url or None, + target=settings.target, + force=settings.force, + ) + + +def _download_asset_paths(xray_dir: Path) -> tuple[Path, ...]: + return tuple(xray_dir / name for name in DOWNLOAD_ASSET_FILES) + + +def _remove_known_paths(paths: list[Path]) -> ClearResult: + removed: list[Path] = [] + missing: list[Path] = [] + for path in dict.fromkeys(paths): + if not path.exists() and not path.is_symlink(): + missing.append(path) + continue + if path.is_dir() and not path.is_symlink(): + shutil.rmtree(path) + else: + path.unlink() + removed.append(path) + return ClearResult(removed=tuple(removed), missing=tuple(missing)) diff --git a/pyxray/libs/xray_asset_settings.py b/pyxray/libs/xray_asset_settings.py index 529ed34..82344f2 100644 --- a/pyxray/libs/xray_asset_settings.py +++ b/pyxray/libs/xray_asset_settings.py @@ -8,7 +8,7 @@ from typing import Any import tomlkit -from pyxray.libs.xray_assets import DEFAULT_VERSION +from pyxray.libs.xray_assets import default_xray_version @dataclass(slots=True) @@ -16,7 +16,7 @@ class XrayAssetSettings: """Xray 资源下载页面的可持久化设置。""" directory: str = "data/xray" - version: str = DEFAULT_VERSION + version: str = "" archive_url: str = "" geoip_url: str = "" geosite_url: str = "" @@ -51,7 +51,7 @@ class XrayAssetSettingsStore: def load(self) -> XrayAssetSettings: if not self.path.exists(): - return XrayAssetSettings(directory=self.default_directory) + return XrayAssetSettings(directory=self.default_directory, version=default_xray_version()) values = dict(tomlkit.parse(self.path.read_text(encoding="utf-8"))) if "directory" not in values: values["directory"] = self.default_directory diff --git a/pyxray/libs/xray_assets.py b/pyxray/libs/xray_assets.py index 14f618c..bbfc219 100644 --- a/pyxray/libs/xray_assets.py +++ b/pyxray/libs/xray_assets.py @@ -1,7 +1,9 @@ from __future__ import annotations import os +import platform import stat +import json import urllib.parse import urllib.request import zipfile @@ -12,13 +14,17 @@ from typing import Callable OFFICIAL_RELEASE_BASE = "https://github.com/XTLS/Xray-core/releases/download" +OFFICIAL_LATEST_RELEASE_API = "https://api.github.com/repos/XTLS/Xray-core/releases/latest" DEFAULT_VERSION = "v26.5.9" DEFAULT_ARCHIVE_NAME = "Xray-linux-64.zip" -REQUIRED_FILES = ("xray", "geoip.dat", "geosite.dat") +REQUIRED_DATA_FILES = ("geoip.dat", "geosite.dat") ASSET_TARGETS = ("all", "xray", "geoip", "geosite") Downloader = Callable[[str], bytes] DownloadProgress = Callable[[str, int, int | None], None] +VersionFetcher = Callable[[str, float], str] + +_DEFAULT_VERSION_CACHE: str | None = None @dataclass(frozen=True, slots=True) @@ -59,10 +65,57 @@ class XrayAssetStatus: return tuple(name for name, exists in self.files.items() if not exists) -def official_archive_url(version: str = DEFAULT_VERSION, archive_name: str = DEFAULT_ARCHIVE_NAME) -> str: +def xray_executable_name(os_name: str | None = None) -> str: + """返回当前平台的 Xray 可执行文件名。""" + + return "xray.exe" if (os_name or os.name) == "nt" else "xray" + + +def required_files(os_name: str | None = None) -> tuple[str, ...]: + """返回当前平台运行 Xray 所需的核心文件。""" + + return (xray_executable_name(os_name), *REQUIRED_DATA_FILES) + + +def default_archive_name(os_name: str | None = None, machine: str | None = None) -> str: + """返回当前平台默认使用的 Xray-core release zip 文件名。""" + + resolved_os = os_name or os.name + resolved_machine = (machine or platform.machine()).lower() + is_arm64 = resolved_machine in {"arm64", "aarch64"} + if resolved_os == "nt": + return "Xray-windows-arm64-v8a.zip" if is_arm64 else "Xray-windows-64.zip" + return "Xray-linux-arm64-v8a.zip" if is_arm64 else DEFAULT_ARCHIVE_NAME + + +def official_archive_url(version: str = DEFAULT_VERSION, archive_name: str | None = None) -> str: """返回官方 Xray-core release zip 下载地址。""" - return f"{OFFICIAL_RELEASE_BASE}/{version}/{archive_name}" + return f"{OFFICIAL_RELEASE_BASE}/{version}/{archive_name or default_archive_name()}" + + +def latest_xray_version(*, timeout: float = 5.0, fetcher: VersionFetcher | None = None) -> str: + """从 GitHub release API 获取最新 Xray-core 版本。""" + + payload = (fetcher or _fetch_url_text)(OFFICIAL_LATEST_RELEASE_API, timeout) + values = json.loads(payload) + tag = str(values.get("tag_name") or "").strip() + if not tag: + raise ValueError("latest Xray release response has no tag_name") + return tag + + +def default_xray_version(*, timeout: float = 5.0) -> str: + """返回默认 Xray 版本；优先远程最新版本，失败时回退到内置版本。""" + + global _DEFAULT_VERSION_CACHE + if _DEFAULT_VERSION_CACHE: + return _DEFAULT_VERSION_CACHE + try: + _DEFAULT_VERSION_CACHE = latest_xray_version(timeout=timeout) + except Exception: # noqa: BLE001 + _DEFAULT_VERSION_CACHE = DEFAULT_VERSION + return _DEFAULT_VERSION_CACHE def check_xray_assets(directory: str | Path) -> XrayAssetStatus: @@ -71,7 +124,7 @@ def check_xray_assets(directory: str | Path) -> XrayAssetStatus: directory = Path(directory) return XrayAssetStatus( directory=directory, - files={name: (directory / name).exists() for name in REQUIRED_FILES}, + files={name: (directory / name).exists() for name in required_files()}, ) @@ -116,17 +169,18 @@ def ensure_xray_assets( if target not in ASSET_TARGETS: raise ValueError(f"unsupported xray asset target: {target}") - xray = directory / "xray" + xray = directory / xray_executable_name() geoip = directory / "geoip.dat" geosite = directory / "geosite.dat" downloaded: list[str] = [] skipped: list[str] = [] requested = _requested_files(target) + xray_name = xray_executable_name() archive_names = { name for name in requested - if name == "xray" + if name == xray_name or (name == "geoip.dat" and geoip_url is None) or (name == "geosite.dat" and geosite_url is None) } @@ -174,6 +228,11 @@ def download_bytes(url: str, *, proxy_url: str | None = None) -> bytes: return response.read() +def _fetch_url_text(url: str, timeout: float) -> str: + with urllib.request.urlopen(url, timeout=timeout) as response: # noqa: S310 + return response.read().decode("utf-8") + + def download_bytes_stream( url: str, progress: DownloadProgress, @@ -250,9 +309,9 @@ def _requested_files(target: str) -> tuple[str, ...]: """把用户选择的下载目标转换为实际文件名。""" if target == "all": - return REQUIRED_FILES + return required_files() if target == "xray": - return ("xray",) + return (xray_executable_name(),) if target == "geoip": return ("geoip.dat",) if target == "geosite": diff --git a/pyxray/web/dashboard.py b/pyxray/web/dashboard.py index be4d619..969dad2 100644 --- a/pyxray/web/dashboard.py +++ b/pyxray/web/dashboard.py @@ -4,7 +4,7 @@ from dataclasses import asdict from flask import Blueprint, Flask, current_app, render_template -from pyxray.libs.xray_assets import DEFAULT_VERSION, check_xray_assets, official_archive_url +from pyxray.libs.xray_assets import check_xray_assets, default_xray_version, official_archive_url from pyxray.libs.xray_config.store import dump_settings_toml from pyxray.web.nodes import get_node_manager from pyxray.web.xray_assets import asset_form_from_settings, get_asset_settings_store @@ -42,7 +42,7 @@ def index(): # noqa: ANN202 status=asdict(status), ready=status.ready, missing=status.missing, - official_url=official_archive_url(form["version"] or DEFAULT_VERSION), + official_url=official_archive_url(form["version"] or default_xray_version()), nodes=nodes, selected_id=selected_id, selected_name=selected_node.name if selected_node is not None else "", diff --git a/pyxray/web/server.py b/pyxray/web/server.py index 274741b..c5fa44e 100644 --- a/pyxray/web/server.py +++ b/pyxray/web/server.py @@ -11,6 +11,7 @@ from pathlib import Path from flask import Flask from pyxray import __version__ +from pyxray.libs.app_data import resolve_app_data_paths from pyxray.web.dashboard import register_dashboard from pyxray.web.jobs import init_job_store from pyxray.web.nodes import register_nodes @@ -28,13 +29,13 @@ def create_app( """创建 pyxray Web 应用。""" app = Flask(__name__) - data_dir = Path(default_data_dir) if default_data_dir is not None else _default_data_dir(default_xray_dir) - config_path = data_dir / "config.json" + paths = resolve_app_data_paths(default_xray_dir, data_dir=default_data_dir) + config_path = paths.generated_config init_job_store(app, run_sync=run_jobs_sync) - register_xray_assets(app, default_xray_dir, data_dir / "download.toml") - register_nodes(app, data_dir / "nodes.toml") - register_xray_config(app, data_dir / "settings.toml", config_path) - register_xray_service(app, xray_dir=default_xray_dir, config_path=config_path, log_path=data_dir / "xray.log") + register_xray_assets(app, paths.xray_dir, paths.download_settings) + register_nodes(app, paths.nodes) + register_xray_config(app, paths.settings, config_path) + register_xray_service(app, xray_dir=paths.xray_dir, config_path=config_path, log_path=paths.log) _bind_xray_lifecycle(app) register_dashboard(app) return app @@ -51,13 +52,6 @@ def run_web(host: str, port: int, default_xray_dir: str | Path = "data/xray") -> create_app(default_xray_dir).run(host=host, port=port) -def _default_data_dir(default_xray_dir: str | Path) -> Path: - """根据资源目录推导默认数据目录。""" - - path = Path(default_xray_dir) - return path.parent if path.name == "xray" else path - - def _print_listen_urls(host: str, port: int) -> None: if host in {"0.0.0.0", "::"}: print(f" * Pyxray URL: http://127.0.0.1:{port}", flush=True) diff --git a/pyxray/web/xray_assets.py b/pyxray/web/xray_assets.py index ee4ff3e..8aa18d7 100644 --- a/pyxray/web/xray_assets.py +++ b/pyxray/web/xray_assets.py @@ -7,6 +7,7 @@ from flask import Blueprint, Flask, current_app, jsonify, request from pyxray.libs.xray_assets import ( DEFAULT_VERSION, check_xray_assets, + default_xray_version, download_bytes_stream, ensure_xray_assets, ) @@ -78,7 +79,7 @@ def default_asset_form(directory: str) -> dict[str, str]: return { "directory": directory, - "version": DEFAULT_VERSION, + "version": default_xray_version(), "archive_url": "", "geoip_url": "", "geosite_url": "", @@ -105,7 +106,7 @@ def _form_values() -> dict[str, str]: def _settings_from_form(form: dict[str, str]) -> XrayAssetSettings: return XrayAssetSettings( directory=form["directory"], - version=form["version"] or DEFAULT_VERSION, + version=form["version"] or default_xray_version(), archive_url=form["archive_url"], geoip_url=form["geoip_url"], geosite_url=form["geosite_url"], diff --git a/tests/libs/test_xray_assets.py b/tests/libs/test_xray_assets.py index d1bb378..9ffe58a 100644 --- a/tests/libs/test_xray_assets.py +++ b/tests/libs/test_xray_assets.py @@ -6,7 +6,17 @@ from urllib.response import addinfourl import pytest -from pyxray.libs.xray_assets import DEFAULT_VERSION, download_bytes, download_bytes_stream, ensure_xray_assets, official_archive_url +from pyxray.libs.xray_assets import ( + DEFAULT_VERSION, + default_archive_name, + default_xray_version, + download_bytes, + download_bytes_stream, + ensure_xray_assets, + latest_xray_version, + official_archive_url, + required_files, +) def _zip_bytes(files: dict[str, bytes]) -> bytes: @@ -19,13 +29,46 @@ def _zip_bytes(files: dict[str, bytes]) -> bytes: def test_official_archive_url_defaults_to_xray_core_v26_5_9() -> None: assert DEFAULT_VERSION == "v26.5.9" - assert official_archive_url() == "https://github.com/XTLS/Xray-core/releases/download/v26.5.9/Xray-linux-64.zip" + assert official_archive_url(archive_name="Xray-linux-64.zip") == ( + "https://github.com/XTLS/Xray-core/releases/download/v26.5.9/Xray-linux-64.zip" + ) + + +def test_default_archive_name_is_platform_specific() -> None: + assert default_archive_name(os_name="posix", machine="x86_64") == "Xray-linux-64.zip" + assert default_archive_name(os_name="nt", machine="AMD64") == "Xray-windows-64.zip" + assert default_archive_name(os_name="nt", machine="ARM64") == "Xray-windows-arm64-v8a.zip" + + +def test_required_files_are_platform_specific() -> None: + assert required_files(os_name="posix") == ("xray", "geoip.dat", "geosite.dat") + assert required_files(os_name="nt") == ("xray.exe", "geoip.dat", "geosite.dat") + + +def test_latest_xray_version_reads_github_release_tag() -> None: + def fetcher(url: str, timeout: float) -> str: + assert url == "https://api.github.com/repos/XTLS/Xray-core/releases/latest" + assert timeout == 5.0 + return '{"tag_name": "v99.1.2"}' + + assert latest_xray_version(fetcher=fetcher) == "v99.1.2" + + +def test_default_xray_version_falls_back_to_pinned_version(monkeypatch) -> None: # noqa: ANN001 + monkeypatch.setattr("pyxray.libs.xray_assets._DEFAULT_VERSION_CACHE", None) + monkeypatch.setattr( + "pyxray.libs.xray_assets.latest_xray_version", + lambda *, timeout: (_ for _ in ()).throw(TimeoutError("slow")), + ) + + assert default_xray_version(timeout=5.0) == DEFAULT_VERSION def test_ensure_xray_assets_extracts_official_archive_files(tmp_path) -> None: # noqa: ANN001 + xray_name = required_files()[0] archive = _zip_bytes( { - "xray": b"bin", + xray_name: b"bin", "geoip.dat": b"geoip", "geosite.dat": b"geosite", "README.md": b"ignored", @@ -42,13 +85,13 @@ def test_ensure_xray_assets_extracts_official_archive_files(tmp_path) -> None: assert result.ready is True assert result.downloaded == ("archive",) assert calls == [official_archive_url()] - assert (tmp_path / "xray").read_bytes() == b"bin" + assert (tmp_path / xray_name).read_bytes() == b"bin" assert (tmp_path / "geoip.dat").read_bytes() == b"geoip" assert (tmp_path / "geosite.dat").read_bytes() == b"geosite" def test_version_and_archive_url_can_be_overridden(tmp_path) -> None: # noqa: ANN001 - archive = _zip_bytes({"xray": b"bin", "geoip.dat": b"geoip", "geosite.dat": b"geosite"}) + archive = _zip_bytes({required_files()[0]: b"bin", "geoip.dat": b"geoip", "geosite.dat": b"geosite"}) calls = [] def downloader(url: str) -> bytes: @@ -61,7 +104,8 @@ def test_version_and_archive_url_can_be_overridden(tmp_path) -> None: # noqa: A def test_dat_urls_override_archive_dat_files(tmp_path) -> None: # noqa: ANN001 - archive = _zip_bytes({"xray": b"bin", "geoip.dat": b"old-geoip", "geosite.dat": b"old-geosite"}) + xray_name = required_files()[0] + archive = _zip_bytes({xray_name: b"bin", "geoip.dat": b"old-geoip", "geosite.dat": b"old-geosite"}) payloads = { official_archive_url(): archive, "https://mirror.invalid/geoip.dat": b"new-geoip", @@ -76,13 +120,14 @@ def test_dat_urls_override_archive_dat_files(tmp_path) -> None: # noqa: ANN001 ) assert result.downloaded == ("archive", "geoip.dat", "geosite.dat") - assert (tmp_path / "xray").read_bytes() == b"bin" + assert (tmp_path / xray_name).read_bytes() == b"bin" assert (tmp_path / "geoip.dat").read_bytes() == b"new-geoip" assert (tmp_path / "geosite.dat").read_bytes() == b"new-geosite" def test_existing_files_skip_download(tmp_path) -> None: # noqa: ANN001 - (tmp_path / "xray").write_bytes(b"bin") + xray_name = required_files()[0] + (tmp_path / xray_name).write_bytes(b"bin") (tmp_path / "geoip.dat").write_bytes(b"geoip") (tmp_path / "geosite.dat").write_bytes(b"geosite") @@ -90,7 +135,7 @@ def test_existing_files_skip_download(tmp_path) -> None: # noqa: ANN001 assert result.ready is True assert result.downloaded == () - assert result.skipped == ("xray", "geoip.dat", "geosite.dat") + assert result.skipped == (xray_name, "geoip.dat", "geosite.dat") def test_force_redownloads_selected_existing_file(tmp_path) -> None: # noqa: ANN001 @@ -116,7 +161,7 @@ def test_force_redownloads_selected_existing_file(tmp_path) -> None: # noqa: AN def test_missing_required_file_raises_after_bad_archive(tmp_path) -> None: # noqa: ANN001 - archive = _zip_bytes({"xray": b"bin", "geoip.dat": b"geoip"}) + archive = _zip_bytes({required_files()[0]: b"bin", "geoip.dat": b"geoip"}) with pytest.raises(FileNotFoundError, match="geosite.dat"): ensure_xray_assets(tmp_path, downloader=lambda url: archive) diff --git a/tests/libs/test_xray_runtime.py b/tests/libs/test_xray_runtime.py index b2ee6fb..a2c9d8d 100644 --- a/tests/libs/test_xray_runtime.py +++ b/tests/libs/test_xray_runtime.py @@ -2,8 +2,12 @@ from __future__ import annotations import subprocess import socket +import os +import shutil +import sys from pathlib import Path +from pyxray.libs.xray_assets import xray_executable_name from pyxray.libs.xray_config import XrayConfigSettings, write_transparent_rule_files from pyxray.libs.xray_runtime import XrayServiceManager from pyxray.libs.xray_transparent_runtime import TransparentRuntime @@ -58,9 +62,7 @@ def test_transparent_runtime_auto_backend_falls_back_to_nft_when_iptables_setup_ def test_xray_service_manager_reports_inbound_port_conflict(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8") - xray.chmod(0o755) + _write_fake_xray(tmp_path) listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) listener.bind(("127.0.0.1", 0)) listener.listen(1) @@ -101,3 +103,17 @@ def _executor(commands: list[list[str]], failures: set[str] | None = None): ) return execute + + +def _write_fake_xray(directory: Path) -> Path: + xray = directory / xray_executable_name() + if os.name == "nt": + try: + os.link(sys.executable, xray) + except OSError: + shutil.copy2(sys.executable, xray) + (directory / "run").write_text("import time\ntime.sleep(30)\n", encoding="utf-8") + return xray + xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8") + xray.chmod(0o755) + return xray diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..3474e45 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +from pathlib import Path + +from pyxray import cli +from pyxray.libs.xray_assets import XrayAssets + + +def test_no_subcommand_defaults_to_web(monkeypatch) -> None: # noqa: ANN001 + captured = {} + + def fake_run_web(host: str, port: int, xray_dir: str) -> None: + captured.update({"host": host, "port": port, "xray_dir": xray_dir}) + + monkeypatch.setattr(cli, "run_web", fake_run_web) + + cli.main([]) + + assert captured == {"host": "0.0.0.0", "port": 8000, "xray_dir": "data/xray"} + + +def test_web_subcommand_still_uses_web_runner(monkeypatch) -> None: # noqa: ANN001 + captured = {} + + def fake_run_web(host: str, port: int, xray_dir: str) -> None: + captured.update({"host": host, "port": port, "xray_dir": xray_dir}) + + monkeypatch.setattr(cli, "run_web", fake_run_web) + + cli.main(["web", "--host", "127.0.0.1", "--port", "9000", "--xray-dir", "runtime/xray"]) + + assert captured == {"host": "127.0.0.1", "port": 9000, "xray_dir": "runtime/xray"} + + +def test_configs_download_prints_download_toml(tmp_path: Path, capsys, monkeypatch) -> None: # noqa: ANN001 + monkeypatch.chdir(tmp_path) + data = tmp_path / "data" + data.mkdir() + (data / "download.toml").write_text('version = "v1.2.3"\n', encoding="utf-8") + + cli.main(["configs", "--download", "--xray-dir", "data/xray"]) + + assert capsys.readouterr().out == 'version = "v1.2.3"\n' + + +def test_clear_download_removes_only_download_settings_and_known_assets(tmp_path: Path, monkeypatch) -> None: # noqa: ANN001 + monkeypatch.chdir(tmp_path) + data = tmp_path / "data" + xray_dir = data / "xray" + xray_dir.mkdir(parents=True) + (data / "download.toml").write_text("download", encoding="utf-8") + (data / "nodes.toml").write_text("nodes", encoding="utf-8") + (xray_dir / "xray").write_bytes(b"xray") + (xray_dir / "xray.exe").write_bytes(b"xray.exe") + (xray_dir / "geoip.dat").write_bytes(b"geoip") + (xray_dir / "geosite.dat").write_bytes(b"geosite") + (xray_dir / "user-file.txt").write_text("keep", encoding="utf-8") + + cli.main(["clear", "--download", "--xray-dir", "data/xray"]) + + assert not (data / "download.toml").exists() + assert not (xray_dir / "xray").exists() + assert not (xray_dir / "xray.exe").exists() + assert not (xray_dir / "geoip.dat").exists() + assert not (xray_dir / "geosite.dat").exists() + assert (data / "nodes.toml").exists() + assert (xray_dir / "user-file.txt").exists() + + +def test_clear_all_removes_known_data_and_keeps_unrelated_files(tmp_path: Path, monkeypatch) -> None: # noqa: ANN001 + monkeypatch.chdir(tmp_path) + data = tmp_path / "data" + xray_dir = data / "xray" + transparent_dir = data / "transparent" + transparent_dir.mkdir(parents=True) + xray_dir.mkdir() + for name in ("download.toml", "nodes.toml", "settings.toml", "config.json", "service-state.json", "xray.log"): + (data / name).write_text(name, encoding="utf-8") + (transparent_dir / "transparent-iptables-setup.sh").write_text("script", encoding="utf-8") + (xray_dir / "xray").write_bytes(b"xray") + (xray_dir / "geoip.dat").write_bytes(b"geoip") + (data / "notes.txt").write_text("keep", encoding="utf-8") + (xray_dir / "custom.dat").write_text("keep", encoding="utf-8") + + cli.main(["clear", "--all", "--xray-dir", "data/xray"]) + + for name in ("download.toml", "nodes.toml", "settings.toml", "config.json", "service-state.json", "xray.log"): + assert not (data / name).exists() + assert not transparent_dir.exists() + assert not (xray_dir / "xray").exists() + assert not (xray_dir / "geoip.dat").exists() + assert (data / "notes.txt").exists() + assert (xray_dir / "custom.dat").exists() + + +def test_download_command_persists_settings_and_reuses_ensure(monkeypatch, tmp_path: Path) -> None: # noqa: ANN001 + monkeypatch.chdir(tmp_path) + captured = {} + + def fake_ensure_xray_assets(directory, **options): # noqa: ANN001 + captured["directory"] = directory + captured["options"] = options + path = Path(directory) + path.mkdir(parents=True, exist_ok=True) + xray = path / "xray" + xray.write_bytes(b"xray") + geoip = path / "geoip.dat" + geoip.write_bytes(b"geoip") + geosite = path / "geosite.dat" + geosite.write_bytes(b"geosite") + return XrayAssets(directory=path, xray=xray, geoip=geoip, geosite=geosite, downloaded=("geoip.dat",)) + + monkeypatch.setattr("pyxray.libs.app_data.ensure_xray_assets", fake_ensure_xray_assets) + + cli.main( + [ + "download", + "--target", + "geoip", + "--directory", + "data/xray", + "--version", + "v1.2.3", + "--geoip-url", + "https://mirror.example.invalid/geoip.dat", + "--force", + ] + ) + + assert captured["directory"] == "data/xray" + assert captured["options"]["target"] == "geoip" + assert captured["options"]["version"] == "v1.2.3" + assert captured["options"]["geoip_url"] == "https://mirror.example.invalid/geoip.dat" + assert captured["options"]["force"] is True + content = (tmp_path / "data" / "download.toml").read_text(encoding="utf-8") + assert 'directory = "data/xray"' in content + assert 'target = "geoip"' in content + assert 'version = "v1.2.3"' in content diff --git a/tests/web/test_xray_assets_web.py b/tests/web/test_xray_assets_web.py index af3a676..2cd0704 100644 --- a/tests/web/test_xray_assets_web.py +++ b/tests/web/test_xray_assets_web.py @@ -3,16 +3,18 @@ from __future__ import annotations import base64 import json import os +import shutil import subprocess +import sys import time from pathlib import Path -from pyxray.libs.xray_assets import XrayAssets +from pyxray.libs.xray_assets import XrayAssets, xray_executable_name from pyxray.web.server import create_app def test_index_shows_asset_status(tmp_path: Path) -> None: - (tmp_path / "xray").write_bytes(b"bin") + (tmp_path / xray_executable_name()).write_bytes(b"bin") app = create_app(tmp_path) client = app.test_client() @@ -42,12 +44,13 @@ def test_ensure_api_uses_form_values(monkeypatch, tmp_path: Path) -> None: # no captured["directory"] = directory captured["options"] = options Path(directory).mkdir(parents=True, exist_ok=True) - (Path(directory) / "xray").write_bytes(b"bin") + xray = Path(directory) / xray_executable_name() + xray.write_bytes(b"bin") (Path(directory) / "geoip.dat").write_bytes(b"geoip") (Path(directory) / "geosite.dat").write_bytes(b"geosite") return XrayAssets( directory=Path(directory), - xray=Path(directory) / "xray", + xray=xray, geoip=Path(directory) / "geoip.dat", geosite=Path(directory) / "geosite.dat", downloaded=("archive",), @@ -121,6 +124,17 @@ def test_asset_settings_api_persists_download_form_values(tmp_path: Path) -> Non assert "v9.9.9" in body +def test_asset_settings_default_version_uses_latest_release(monkeypatch, tmp_path: Path) -> None: # noqa: ANN001 + monkeypatch.setattr("pyxray.libs.xray_assets._DEFAULT_VERSION_CACHE", None) + monkeypatch.setattr("pyxray.libs.xray_assets.latest_xray_version", lambda *, timeout: "v99.9.9") + app = create_app(tmp_path) + client = app.test_client() + + payload = client.get("/api/xray/assets/settings").get_json() + + assert payload["version"] == "v99.9.9" + + def test_job_records_real_download_percent(monkeypatch, tmp_path: Path) -> None: # noqa: ANN001 def fake_download_bytes_stream(url, progress, **options): # noqa: ANN001 progress(url, 0, 10) @@ -131,12 +145,13 @@ def test_job_records_real_download_percent(monkeypatch, tmp_path: Path) -> None: def fake_ensure_xray_assets(directory, *, downloader, **options): # noqa: ANN001, ARG001 downloader("https://mirror.invalid/xray.zip") Path(directory).mkdir(parents=True, exist_ok=True) - (Path(directory) / "xray").write_bytes(b"bin") + xray = Path(directory) / xray_executable_name() + xray.write_bytes(b"bin") (Path(directory) / "geoip.dat").write_bytes(b"geoip") (Path(directory) / "geosite.dat").write_bytes(b"geosite") return XrayAssets( directory=Path(directory), - xray=Path(directory) / "xray", + xray=xray, geoip=Path(directory) / "geoip.dat", geosite=Path(directory) / "geosite.dat", downloaded=("archive",), @@ -173,9 +188,7 @@ def test_cancel_job_api_marks_job_cancel_requested(tmp_path: Path) -> None: def test_xray_service_api_starts_stops_and_reads_logs(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\necho xray-started\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path, stdout="xray-started") app = create_app(tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -199,9 +212,7 @@ def test_xray_service_api_starts_stops_and_reads_logs(tmp_path: Path) -> None: def test_xray_service_restores_desired_running_state_on_app_start(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\necho restored-start\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path, stdout="restored-start") app = create_app(tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -228,9 +239,7 @@ def test_xray_service_restores_desired_running_state_on_app_start(tmp_path: Path def test_xray_service_log_forwarder_flushes_line_output_quickly(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\necho first-line\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path, stdout="first-line") app = create_app(tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -247,9 +256,7 @@ def test_xray_service_log_forwarder_flushes_line_output_quickly(tmp_path: Path) def test_xray_service_start_regenerates_config_from_saved_settings(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path) app = create_app(tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -268,9 +275,7 @@ def test_xray_service_start_regenerates_config_from_saved_settings(tmp_path: Pat def test_xray_service_applies_transparent_rules_on_start_and_cleans_on_stop(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path) app = create_app(tmp_path) commands: list[str] = [] app.extensions["pyxray_transparent_runtime"].executor = _recording_executor(commands) @@ -307,9 +312,7 @@ def test_xray_service_applies_transparent_rules_on_start_and_cleans_on_stop(tmp_ def test_xray_service_rolls_back_when_transparent_setup_fails(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path) app = create_app(tmp_path) commands: list[str] = [] app.extensions["pyxray_transparent_runtime"].executor = _recording_executor( @@ -344,9 +347,7 @@ def test_xray_service_rolls_back_when_transparent_setup_fails(tmp_path: Path) -> def test_xray_service_shutdown_stops_managed_process(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path) app = create_app(tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -366,9 +367,7 @@ def test_xray_service_uses_absolute_paths_when_app_created_with_relative_xray_di monkeypatch.chdir(tmp_path) xray_dir = tmp_path / "data" / "xray" xray_dir.mkdir(parents=True) - xray = xray_dir / "xray" - xray.write_text("#!/bin/sh\necho relative-started\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + xray = _write_fake_xray(xray_dir, stdout="relative-started") app = create_app("data/xray") client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -385,9 +384,7 @@ def test_xray_service_uses_absolute_paths_when_app_created_with_relative_xray_di def test_xray_service_api_reports_missing_config(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path) app = create_app(tmp_path) client = app.test_client() @@ -398,9 +395,7 @@ def test_xray_service_api_reports_missing_config(tmp_path: Path) -> None: def test_xray_service_api_records_immediate_start_failure_output(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\necho 'bind: permission denied' >&2\nexit 23\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path, stderr="bind: permission denied", exit_code=23, sleep_seconds=0) app = create_app(tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -488,10 +483,8 @@ def test_xray_service_prefers_persisted_download_directory(tmp_path: Path) -> No preferred_dir = tmp_path / "download-xray" default_dir.mkdir() preferred_dir.mkdir() - (default_dir / "xray").write_text("#!/bin/sh\necho default-xray\nsleep 30\n", encoding="utf-8") - (preferred_dir / "xray").write_text("#!/bin/sh\necho preferred-xray\nsleep 30\n", encoding="utf-8") - os.chmod(default_dir / "xray", 0o755) - os.chmod(preferred_dir / "xray", 0o755) + _write_fake_xray(default_dir, stdout="default-xray") + preferred_xray = _write_fake_xray(preferred_dir, stdout="preferred-xray") app = create_app(default_dir, default_data_dir=tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -503,9 +496,9 @@ def test_xray_service_prefers_persisted_download_directory(tmp_path: Path) -> No client.post("/api/xray/service/stop") logs = client.get("/api/xray/service/logs").get_json()["content"] - assert started["xray"] == str(preferred_dir / "xray") + assert started["xray"] == str(preferred_xray) assert started["xray_dir"] == str(preferred_dir) - assert str(preferred_dir / "xray") in logs + assert str(preferred_xray) in logs def test_xray_service_falls_back_to_default_directory_when_saved_directory_has_no_xray(tmp_path: Path) -> None: @@ -513,8 +506,7 @@ def test_xray_service_falls_back_to_default_directory_when_saved_directory_has_n preferred_dir = tmp_path / "download-xray" default_dir.mkdir() preferred_dir.mkdir() - (default_dir / "xray").write_text("#!/bin/sh\necho default-xray\nsleep 30\n", encoding="utf-8") - os.chmod(default_dir / "xray", 0o755) + default_xray = _write_fake_xray(default_dir, stdout="default-xray") app = create_app(default_dir, default_data_dir=tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -526,10 +518,10 @@ def test_xray_service_falls_back_to_default_directory_when_saved_directory_has_n client.post("/api/xray/service/stop") logs = client.get("/api/xray/service/logs").get_json()["content"] - assert started["xray"] == str(default_dir / "xray") + assert started["xray"] == str(default_xray) assert started["xray_dir"] == str(default_dir) assert started["fallback_xray_dir"] == str(default_dir) - assert str(default_dir / "xray") in logs + assert str(default_xray) in logs def test_nodes_api_imports_lists_selects_and_deletes_node(tmp_path: Path) -> None: @@ -554,9 +546,7 @@ def test_nodes_api_imports_lists_selects_and_deletes_node(tmp_path: Path) -> Non def test_selecting_node_restarts_running_xray(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\necho started-$@\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path, stdout="started", echo_args=True) app = create_app(tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": "\n".join([_ss_link("one", "one"), _ss_link("two", "two")])}) @@ -603,9 +593,7 @@ def test_xray_config_api_saves_settings_and_generates_config(tmp_path: Path) -> def test_saving_settings_restarts_running_xray(tmp_path: Path) -> None: - xray = tmp_path / "xray" - xray.write_text("#!/bin/sh\necho settings-started\nsleep 30\n", encoding="utf-8") - os.chmod(xray, 0o755) + _write_fake_xray(tmp_path, stdout="settings-started") app = create_app(tmp_path) client = app.test_client() client.post("/api/nodes/import", data={"links": _ss_link("secret", "ss-node")}) @@ -838,6 +826,51 @@ def _ss_link(password: str, name: str) -> str: return f"ss://{user}@ss.example.net:8388#{name}" +def _write_fake_xray( + directory: Path, + *, + stdout: str = "", + stderr: str = "", + exit_code: int = 0, + sleep_seconds: float = 30, + echo_args: bool = False, +) -> Path: + xray = directory / xray_executable_name() + code = _fake_xray_code(stdout=stdout, stderr=stderr, exit_code=exit_code, sleep_seconds=sleep_seconds, echo_args=echo_args) + if os.name == "nt": + try: + os.link(sys.executable, xray) + except OSError: + shutil.copy2(sys.executable, xray) + (directory / "run").write_text(code, encoding="utf-8") + return xray + xray.write_text(f"#!{sys.executable}\n{code}", encoding="utf-8") + xray.chmod(0o755) + return xray + + +def _fake_xray_code( + *, + stdout: str, + stderr: str, + exit_code: int, + sleep_seconds: float, + echo_args: bool, +) -> str: + lines = ["import sys", "import time"] + if stdout: + if echo_args: + lines.append(f"print({stdout!r} + ' ' + ' '.join(sys.argv[1:]), flush=True)") + else: + lines.append(f"print({stdout!r}, flush=True)") + if stderr: + lines.append(f"print({stderr!r}, file=sys.stderr, flush=True)") + if sleep_seconds: + lines.append(f"time.sleep({sleep_seconds!r})") + lines.append(f"raise SystemExit({exit_code})") + return "\n".join(lines) + "\n" + + def _recording_executor( commands: list[str], failures: set[str] | None = None, diff --git a/uv.lock b/uv.lock index 0ac4934..cb822ef 100644 --- a/uv.lock +++ b/uv.lock @@ -154,7 +154,7 @@ wheels = [ [[package]] name = "pyxray" -version = "1.0.4" +version = "1.0.5" source = { editable = "." } dependencies = [ { name = "flask" },