diff --git a/src/xskill/cli.py b/src/xskill/cli.py index a314b8e7..8d56cc3e 100644 --- a/src/xskill/cli.py +++ b/src/xskill/cli.py @@ -244,9 +244,12 @@ def _connect_handshake(args, state_path): # 作为 ``claimed_client_id`` 一起发给 server——server 按 (claimed/fingerprint/ # new) 三级判定续用。state 不在 → existing_client_id=None,让 server 按指纹回查。 existing_client_id: str | None = None + existing_pypi_url: str | None = None if state_path.is_file(): try: - existing_client_id = load_client_state(state_path).client_id + existing_state = load_client_state(state_path) + existing_client_id = existing_state.client_id + existing_pypi_url = existing_state.pypi_url except Exception: # state 文件损坏不阻断重连——按"无本地身份"处理,让 server 走指纹回查 # 或新发。损坏的 state 接下来会被新的 save 覆盖。 @@ -269,7 +272,8 @@ def _connect_handshake(args, state_path): print(f"error: 注册失败: {e}", file=sys.stderr) return None state = ClientState(server_url=server_url, client_id=client_id, - join_token=args.token) + join_token=args.token, + pypi_url=args.pypi_url or existing_pypi_url) save_client_state(state, state_path) name_hint = f" (--name={args.name})" if args.name else "" print(f"connected: client_id={client_id} server={server_url}{name_hint}") @@ -337,41 +341,47 @@ def cmd_start(args) -> int: def cmd_update(args) -> int: - """立即检查 PyPI 是否有新版 xskill,有则升级并重启。""" - from xskill.team.client.updater import ( - _current_version, _latest_pypi_version, _restart, - ) + """立即检查是否有新版 xskill,有则升级并重启。 + + 复用 ``AutoUpdater`` 的三级回退链路(公网 PyPI → 已连接过的内网镜像 → + team server wheel),而不是自己另开一条查询/安装逻辑——这样手动 + `xskill update` 和后台每小时自动检查行为完全一致,不会出现"自动更新能 + 绕过公网不可达,手动触发却直接报错"的不一致。 + """ + from xskill.config import get_team_client_state_path + from xskill.team.client.state import load_client_state + from xskill.team.client.updater import AutoUpdater, _current_version + current = _current_version("xskill") if not current: print("error: 无法读取当前版本", file=sys.stderr) return 1 print(f"当前版本: {current}") - print("正在查询 PyPI...") - latest = _latest_pypi_version("xskill") - if not latest: - print("error: 查询 PyPI 失败,请检查网络", file=sys.stderr) - return 1 - try: - from packaging.version import Version - if Version(latest) <= Version(current): - print(f"已是最新版本 ({current})") - return 0 - except Exception: - pass - print(f"发现新版本: {latest},开始升级...") - import subprocess - result = subprocess.run( - [sys.executable, "-m", "pip", "install", "--upgrade", - f"xskill=={latest}", "-i", "https://pypi.org/simple/"], - capture_output=True, text=True, - ) - if result.returncode != 0: - print(f"error: 升级失败:\n{result.stderr.strip() or result.stdout.strip()}", + + kwargs: dict = {} + state_path = get_team_client_state_path() + if state_path.is_file(): + try: + state = load_client_state(state_path) + kwargs.update(server_url=state.server_url, client_id=state.client_id, + join_token=state.join_token) + if state.pypi_url: + kwargs["pypi_url"] = state.pypi_url + except Exception: + pass # state 文件损坏:退化为「只查公网 PyPI」 + + print("正在检查更新(公网 PyPI" + + ("、内网镜像" if kwargs.get("pypi_url") else "") + + ("、team server 回退" if kwargs.get("server_url") else "") + ")...") + + updater = AutoUpdater(**kwargs) + if not updater.run_once(): + print("error: 升级失败:所有可用来源均不可用或安装失败,详情见上方日志", file=sys.stderr) return 1 - print(f"升级到 {latest} 成功,正在重启...") - _restart() - return 0 # 不会到达这里 + # 升级成功时 _restart() 已替换/重启进程,走不到这里;到这里说明本来就是最新版本。 + print(f"已是最新版本 ({current})") + return 0 def cmd_stop(args) -> int: @@ -649,6 +659,13 @@ def build_parser() -> argparse.ArgumentParser: "--no-auto-update", action="store_true", dest="no_auto_update", help="禁用自动更新检查(默认每小时查一次 PyPI,有新版则升级重启)。", ) + p_conn.add_argument( + "--pypi-url", default=None, metavar="URL", + help="内网 PyPI 镜像地址(PEP 503 /simple/ 索引),自动更新会在公网 " + "PyPI 不可达时用它查版本、装包,装 server 回退 wheel 时的依赖也走它。" + "落盘到本地连接信息,后续 `xskill connect` / `xskill update` 复用," + "不用每次都传。默认不设,只用公网 PyPI + server wheel 回退。", + ) p_start = sub.add_parser( "start", help="把 connect 装成后台常驻(开机自启 + 崩溃自愈)", diff --git a/src/xskill/team/client/daemon.py b/src/xskill/team/client/daemon.py index bc006e4a..8c284619 100644 --- a/src/xskill/team/client/daemon.py +++ b/src/xskill/team/client/daemon.py @@ -356,11 +356,12 @@ def _tick(self) -> None: def run_forever(self) -> None: """阻塞循环。先起 collector ingester,再每 poll_interval 跑一轮 _tick。""" - from xskill.team.client.updater import AutoUpdater + from xskill.team.client.updater import AutoUpdater, PUBLIC_PYPI_SIMPLE_URL updater = AutoUpdater( server_url=self.state.server_url, client_id=self.state.client_id, join_token=self.state.join_token, + pypi_url=self.state.pypi_url or PUBLIC_PYPI_SIMPLE_URL, ) if self.auto_update else None if updater: updater.start() diff --git a/src/xskill/team/client/state.py b/src/xskill/team/client/state.py index 77ec8558..df03411d 100644 --- a/src/xskill/team/client/state.py +++ b/src/xskill/team/client/state.py @@ -18,6 +18,7 @@ class ClientState: server_url: str # 形如 http://1.2.3.4:8000 client_id: str join_token: str + pypi_url: str | None = None # 内网 PyPI 镜像,供自动更新回退用;None=只用公网 PyPI def save_client_state(state: ClientState, path: Path | str) -> None: @@ -40,4 +41,5 @@ def load_client_state(path: Path | str) -> ClientState: server_url=data["server_url"], client_id=data["client_id"], join_token=data["join_token"], + pypi_url=data.get("pypi_url"), ) diff --git a/src/xskill/team/client/updater.py b/src/xskill/team/client/updater.py index 8b968a34..a59e0281 100644 --- a/src/xskill/team/client/updater.py +++ b/src/xskill/team/client/updater.py @@ -1,8 +1,15 @@ """updater.py — xskill client 自动更新 TeamClient 跑起来后每隔一段时间(默认 1 小时)查 PyPI,发现新版就升级并重启。 -如果 PyPI 查询或安装失败,且 client 已连接 team server,则读取 server 版本; -server 版本高于本地版本时,下载 server 暴露的 wheel 并安装。 +查询/安装链路三级回退: + +1. 公网 PyPI(JSON API 查版本 + ``pip install -i pypi_url``) +2. 已配置的内网 PyPI 镜像(``pypi_url`` 非公网地址时,``pip index versions`` + 查版本 + 复用同一 ``-i pypi_url`` 安装)——多数内网镜像(devpi/artifactory) + 不提供 pypi.org 的 legacy JSON API,所以查询手段和公网不同,但安装命令相同。 +3. 都不可用,且 client 已连接 team server:读取 server 版本,高于本地时下载 + server 暴露的 wheel 安装——wheel 依赖同样经 ``pypi_url``(配了镜像就用镜像, + 否则退回公网)解析,不会因为装了新依赖又绕回不可达的公网索引。 重启机制 ──────── @@ -13,7 +20,7 @@ ──────── - 包含预发版(a/b/rc),因为内部用 alpha 版本 - 严格大于当前版本才升级,不降级 -- 网络/PyPI/server 故障不会打断主循环 +- 网络/PyPI/镜像/server 故障不会打断主循环 """ from __future__ import annotations @@ -29,6 +36,7 @@ logger = logging.getLogger("xskill.team.client.updater") _PYPI_JSON_URL = "https://pypi.org/pypi/{package}/json" +PUBLIC_PYPI_SIMPLE_URL = "https://pypi.org/simple/" def _team_api_url(server_url: str, path: str) -> str: @@ -73,6 +81,33 @@ def _latest_pypi_version(package: str) -> Optional[str]: return None +def _latest_mirror_version(package: str, pypi_url: str) -> Optional[str]: + """查内网 PyPI 镜像取最新版本(含预发版)。 + + 内网镜像(devpi/artifactory 等)通常不实现 pypi.org 的 legacy JSON API, + 只有标准 PEP 503 ``/simple/`` 索引,所以改用 ``pip index versions`` + (走的正是同一个 ``/simple/`` 索引,公网/镜像通用)。超时/查询失败返回 None。 + """ + try: + result = subprocess.run( + [sys.executable, "-m", "pip", "index", "versions", package, + "--pre", "-i", pypi_url], + capture_output=True, text=True, timeout=20, + ) + from packaging.version import Version + for line in result.stdout.splitlines(): + if not line.startswith("Available versions:"): + continue + versions = [v.strip() for v in line.split(":", 1)[1].split(",") if v.strip()] + parsed = [Version(v) for v in versions] + if parsed: + return str(max(parsed)) + return None + except Exception: + logger.debug("updater: 查镜像 %s 版本失败", pypi_url, exc_info=True) + return None + + def _server_version( server_url: str, join_token: str, @@ -194,7 +229,7 @@ def __init__( self, package: str = "xskill", interval: float = 3600, # 默认 1 小时 - pypi_url: str = "https://pypi.org/simple/", + pypi_url: str = PUBLIC_PYPI_SIMPLE_URL, server_url: str | None = None, client_id: str | None = None, join_token: str | None = None, @@ -221,6 +256,20 @@ def start(self) -> None: def stop(self) -> None: self._stop.set() + def run_once(self) -> bool: + """立即检查一次并在有新版时升级(供 ``xskill update`` 手动触发)。 + + 返回是否成功:已是最新版本 / 升级成功(后者 ``_restart()`` 会直接 + 替换进程,通常不会真的返回到这里)都算 True;所有可用来源都升级 + 失败才是 False。 + """ + return self._check_and_update() + + def _has_mirror(self) -> bool: + return bool(self.pypi_url) and ( + self.pypi_url.rstrip("/") != PUBLIC_PYPI_SIMPLE_URL.rstrip("/") + ) + # ───────────────────────────────────────────────────────────── def _loop(self) -> None: @@ -230,39 +279,42 @@ def _loop(self) -> None: self._check_and_update() self._stop.wait(self.interval) - def _check_and_update(self) -> None: + def _check_and_update(self) -> bool: current_str = _current_version(self.package) if not current_str: logger.debug("updater: 无法读取当前版本,跳过本次检查") - return + return False try: from packaging.version import Version current = Version(current_str) except Exception: logger.debug("updater: 当前版本不可解析: %s", current_str, exc_info=True) - return + return False latest_str = _latest_pypi_version(self.package) + reason = "pypi_query_failed" + if not latest_str and self._has_mirror(): + latest_str = _latest_mirror_version(self.package, self.pypi_url) + reason = "mirror_query_failed" if not latest_str: - self._check_server_fallback(current_str, current, reason="pypi_query_failed") - return + return self._check_server_fallback(current_str, current, reason=reason) try: latest = Version(latest_str) except Exception: - return + return False if latest <= current: logger.debug("updater: 当前版本 %s 已是最新", current_str) - return + return True logger.info("updater: 发现新版本 %s(当前 %s),开始升级...", latest_str, current_str) if self._install(latest_str): _restart() # 升级成功后重启,不会走到这行之后的代码 # (_restart 在 Windows 上 os._exit;Linux 上 execv) - return - self._check_server_fallback(current_str, current, reason="pypi_install_failed") + return True + return self._check_server_fallback(current_str, current, reason="pypi_install_failed") def _install(self, target_version: str) -> bool: """用 pip 升级到指定版本。返回是否成功。""" @@ -284,15 +336,20 @@ def _install(self, target_version: str) -> bool: logger.warning("updater: 执行 pip 失败", exc_info=True) return False - def _check_server_fallback(self, current_str: str, current, *, reason: str) -> None: - """PyPI 不可用时,从 team server 下载同版本 wheel 回退升级。""" + def _check_server_fallback(self, current_str: str, current, *, reason: str) -> bool: + """PyPI/镜像都不可用时,从 team server 下载同版本 wheel 回退升级。 + + 返回值语义与 ``_check_and_update`` 一致:没有更高版本可装(无论是 + 因为没配 server、server 版本不比本地新,还是本来就已最新)算 True; + 明确「有更新但装不上」才是 False。 + """ if not (self.server_url and self.client_id and self.join_token): logger.debug("updater: 无 server 回退配置,跳过(%s)", reason) - return + return False info = _server_version(self.server_url, self.join_token, self.client_id) if not info: - return + return False server_version_str = str(info.get("version") or "") try: @@ -301,16 +358,16 @@ def _check_server_fallback(self, current_str: str, current, *, reason: str) -> N except Exception: logger.debug("updater: server 版本不可解析: %s", server_version_str, exc_info=True) - return + return False if server_version <= current: logger.debug("updater: server 版本 %s 不高于当前版本 %s", server_version_str, current_str) - return + return True if not info.get("wheel_available"): logger.warning("updater: server 版本 %s 可用,但未提供 wheel", server_version_str) - return + return False with tempfile.TemporaryDirectory(prefix="xskill-update-") as td: wheel = _download_server_wheel( @@ -321,17 +378,25 @@ def _check_server_fallback(self, current_str: str, current, *, reason: str) -> N str(info.get("wheel_filename") or ""), ) if wheel is None: - return - logger.info("updater: PyPI 不可用(%s),改用 server wheel 升级到 %s", + return False + logger.info("updater: PyPI/镜像不可用(%s),改用 server wheel 升级到 %s", reason, server_version_str) if self._install_wheel(wheel): _restart() + return True + return False def _install_wheel(self, wheel_path: Path) -> bool: - """用 pip 安装 server 下载的 wheel。返回是否成功。""" + """用 pip 安装 server 下载的 wheel。返回是否成功。 + + 显式传 ``-i self.pypi_url``:wheel 本体来自 server,但它的依赖仍要 + 走一个可达的索引解析——配了内网镜像就用镜像,没配就退回公网 PyPI, + 绝不能静默吃 pip 自己的默认索引配置(那台机器上很可能压根没配)。 + """ cmd = [ sys.executable, "-m", "pip", "install", "--upgrade", str(wheel_path), + "-i", self.pypi_url, "-q", ] try: diff --git a/tests/test_cli_update.py b/tests/test_cli_update.py new file mode 100644 index 00000000..17ed6865 --- /dev/null +++ b/tests/test_cli_update.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from xskill.cli import build_parser, cmd_update + + +def _args(): + return build_parser().parse_args(["update"]) + + +def test_update_no_current_version_errors(monkeypatch, tmp_path): + monkeypatch.setattr("xskill.config.get_team_client_state_path", + lambda: tmp_path / "absent.json") + monkeypatch.setattr( + "xskill.team.client.updater._current_version", lambda package: None) + rc = cmd_update(_args()) + assert rc == 1 + + +def test_update_success_delegates_to_autoupdater_run_once(monkeypatch, tmp_path, capsys): + monkeypatch.setattr("xskill.config.get_team_client_state_path", + lambda: tmp_path / "absent.json") + monkeypatch.setattr( + "xskill.team.client.updater._current_version", lambda package: "1.0.0") + + calls: list[bool] = [] + + class _FakeUpdater: + def __init__(self, **kwargs): + self.kwargs = kwargs + + def run_once(self): + calls.append(True) + return True + + monkeypatch.setattr("xskill.team.client.updater.AutoUpdater", _FakeUpdater) + rc = cmd_update(_args()) + assert rc == 0 + assert calls == [True] + assert "已是最新版本" in capsys.readouterr().out + + +def test_update_all_sources_fail_returns_error(monkeypatch, tmp_path): + monkeypatch.setattr("xskill.config.get_team_client_state_path", + lambda: tmp_path / "absent.json") + monkeypatch.setattr( + "xskill.team.client.updater._current_version", lambda package: "1.0.0") + + class _FakeUpdater: + def __init__(self, **kwargs): + pass + + def run_once(self): + return False + + monkeypatch.setattr("xskill.team.client.updater.AutoUpdater", _FakeUpdater) + rc = cmd_update(_args()) + assert rc == 1 + + +def test_update_reuses_persisted_server_and_pypi_url(monkeypatch, tmp_path): + """已 connect 过的机器:team_client.json 里的 server/token/pypi_url 应 + 自动喂给 AutoUpdater,不需要用户在 `xskill update` 时重新指定。""" + from xskill.team.client.state import ClientState, save_client_state + + state_path = tmp_path / "team_client.json" + save_client_state( + ClientState(server_url="http://server", client_id="cid-1", + join_token="tok", pypi_url="http://mirror/simple"), + state_path, + ) + monkeypatch.setattr("xskill.config.get_team_client_state_path", lambda: state_path) + monkeypatch.setattr( + "xskill.team.client.updater._current_version", lambda package: "1.0.0") + + captured: dict = {} + + class _FakeUpdater: + def __init__(self, **kwargs): + captured.update(kwargs) + + def run_once(self): + return True + + monkeypatch.setattr("xskill.team.client.updater.AutoUpdater", _FakeUpdater) + rc = cmd_update(_args()) + assert rc == 0 + assert captured == { + "server_url": "http://server", + "client_id": "cid-1", + "join_token": "tok", + "pypi_url": "http://mirror/simple", + } diff --git a/tests/test_team_cli_connect.py b/tests/test_team_cli_connect.py index 71cf5a21..126fa96d 100644 --- a/tests/test_team_cli_connect.py +++ b/tests/test_team_cli_connect.py @@ -245,3 +245,49 @@ def install_and_start(self): rc = cmd_connect(args) assert rc == 0 assert ran["forever"] == 1 # 退化成前台阻塞 + + +# ── --pypi-url:内网镜像地址落盘,供自动更新回退用 ── + +def test_pypi_url_flag_parses(): + parser = build_parser() + args = parser.parse_args(["connect", "1.2.3.4:8000", "--token", "t"]) + assert args.pypi_url is None + args2 = parser.parse_args( + ["connect", "1.2.3.4:8000", "--token", "t", + "--pypi-url", "http://mirror/simple"]) + assert args2.pypi_url == "http://mirror/simple" + + +def test_pypi_url_persisted_on_first_connect(tmp_path, monkeypatch): + from xskill.team.client.state import load_client_state + state_path = tmp_path / "team_client.json" + monkeypatch.setattr("xskill.config.get_team_client_state_path", + lambda: state_path) + _install_fakes(monkeypatch) + parser = build_parser() + args = parser.parse_args( + ["connect", "1.2.3.4:8000", "--token", "t", + "--pypi-url", "http://mirror/simple"]) + rc = cmd_connect(args) + assert rc == 0 + assert load_client_state(state_path).pypi_url == "http://mirror/simple" + + +def test_pypi_url_preserved_on_reconnect_without_flag(tmp_path, monkeypatch): + """重连时不重复传 --pypi-url,应沿用上次已存的镜像地址,而不是被清空。""" + from xskill.team.client.state import ClientState, load_client_state, save_client_state + state_path = tmp_path / "team_client.json" + monkeypatch.setattr("xskill.config.get_team_client_state_path", + lambda: state_path) + save_client_state( + ClientState(server_url="http://1.2.3.4:8000", client_id="cid-1", + join_token="old-tok", pypi_url="http://mirror/simple"), + state_path, + ) + _install_fakes(monkeypatch) + parser = build_parser() + args = parser.parse_args(["connect", "1.2.3.4:8000", "--token", "t"]) + rc = cmd_connect(args) + assert rc == 0 + assert load_client_state(state_path).pypi_url == "http://mirror/simple" diff --git a/tests/test_team_client_state.py b/tests/test_team_client_state.py index 97d6cffe..674ad38e 100644 --- a/tests/test_team_client_state.py +++ b/tests/test_team_client_state.py @@ -17,3 +17,23 @@ def test_save_then_load_roundtrip(tmp_path): def test_load_missing_raises(tmp_path): with pytest.raises(FileNotFoundError): load_client_state(tmp_path / "absent.json") + + +def test_save_then_load_roundtrip_with_pypi_url(tmp_path): + p = tmp_path / "team_client.json" + st = ClientState(server_url="http://1.2.3.4:8000", client_id="cid-1", + join_token="tok", pypi_url="http://mirror/simple") + save_client_state(st, p) + back = load_client_state(p) + assert back == st + + +def test_load_old_format_without_pypi_url_defaults_none(tmp_path): + """老版本写的 team_client.json 没有 pypi_url 字段,读取时应默认 None 而不是报错。""" + import json + p = tmp_path / "team_client.json" + p.write_text(json.dumps({ + "server_url": "http://1.2.3.4:8000", "client_id": "cid-1", "join_token": "tok", + })) + back = load_client_state(p) + assert back.pypi_url is None diff --git a/tests/test_team_client_updater.py b/tests/test_team_client_updater.py index c32aa957..6da90f63 100644 --- a/tests/test_team_client_updater.py +++ b/tests/test_team_client_updater.py @@ -3,7 +3,43 @@ from pathlib import Path from xskill.team.client import updater as updater_mod -from xskill.team.client.updater import AutoUpdater +from xskill.team.client.updater import AutoUpdater, _latest_mirror_version + + +def test_latest_mirror_version_parses_pip_index_output(monkeypatch): + class FakeResult: + stdout = ( + "xskill (1.3.0)\n" + "Available versions: 1.3.0, 1.2.0a3, 1.2.0, 1.0.0\n" + " INSTALLED: 1.0.0\n" + ) + stderr = "" + returncode = 0 + + captured_cmds: list[list[str]] = [] + + def fake_run(cmd, **kwargs): + captured_cmds.append(cmd) + return FakeResult() + + monkeypatch.setattr(updater_mod.subprocess, "run", fake_run) + + version = _latest_mirror_version("xskill", "http://mirror/simple") + + assert version == "1.3.0" + cmd = captured_cmds[0] + assert cmd[:5] == [updater_mod.sys.executable, "-m", "pip", "index", "versions"] + assert "-i" in cmd and cmd[cmd.index("-i") + 1] == "http://mirror/simple" + assert "--pre" in cmd + + +def test_latest_mirror_version_returns_none_on_failure(monkeypatch): + def fake_run(cmd, **kwargs): + raise OSError("no route to host") + + monkeypatch.setattr(updater_mod.subprocess, "run", fake_run) + + assert _latest_mirror_version("xskill", "http://mirror/simple") is None def test_pypi_newer_version_uses_pypi_and_skips_server(monkeypatch): @@ -156,3 +192,79 @@ def test_server_fallback_skips_non_newer_server_version(monkeypatch): updater._check_and_update() assert downloaded == [] + + +def test_mirror_tried_before_server_when_pypi_unreachable(monkeypatch): + """公网 PyPI 查询失败但配了内网镜像时,应该先试镜像,不直接跳到 server。""" + installed: list[str] = [] + restarted: list[bool] = [] + + monkeypatch.setattr(updater_mod, "_current_version", lambda package: "1.0.0") + monkeypatch.setattr(updater_mod, "_latest_pypi_version", lambda package: None) + monkeypatch.setattr( + updater_mod, "_latest_mirror_version", + lambda package, pypi_url: "1.3.0" if pypi_url == "http://mirror/simple" else None, + ) + monkeypatch.setattr( + updater_mod, "_server_version", + lambda *a, **k: (_ for _ in ()).throw(AssertionError("server fallback used")), + ) + monkeypatch.setattr(updater_mod, "_restart", lambda: restarted.append(True)) + + updater = AutoUpdater(pypi_url="http://mirror/simple") + monkeypatch.setattr(updater, "_install", lambda version: installed.append(version) or True) + + ok = updater._check_and_update() + + assert installed == ["1.3.0"] + assert restarted == [True] + assert ok is True + + +def test_mirror_not_queried_when_not_configured(monkeypatch): + """没配镜像(pypi_url 仍是公网默认值)时,PyPI 查询失败应直接走 server 回退, + 不应该尝试查一个根本没配置的"镜像"。""" + monkeypatch.setattr(updater_mod, "_current_version", lambda package: "1.0.0") + monkeypatch.setattr(updater_mod, "_latest_pypi_version", lambda package: None) + monkeypatch.setattr( + updater_mod, "_latest_mirror_version", + lambda *a, **k: (_ for _ in ()).throw(AssertionError("mirror query used")), + ) + + updater = AutoUpdater() # 无 server 配置 -> fallback 也是空操作 + ok = updater._check_and_update() + + assert ok is False + + +def test_install_wheel_passes_pypi_url_as_index(monkeypatch): + """server wheel 的依赖必须走一个明确可达的索引(镜像优先),不能吃 pip 默认索引。""" + captured_cmds: list[list[str]] = [] + + class FakeResult: + returncode = 0 + stdout = "" + stderr = "" + + def fake_run(cmd, **kwargs): + captured_cmds.append(cmd) + return FakeResult() + + monkeypatch.setattr(updater_mod.subprocess, "run", fake_run) + + updater = AutoUpdater(pypi_url="http://mirror/simple") + ok = updater._install_wheel(Path("/tmp/xskill-1.3.0-py3-none-any.whl")) + + assert ok is True + assert captured_cmds + cmd = captured_cmds[0] + assert "-i" in cmd + assert cmd[cmd.index("-i") + 1] == "http://mirror/simple" + + +def test_up_to_date_returns_true(monkeypatch): + monkeypatch.setattr(updater_mod, "_current_version", lambda package: "1.0.0") + monkeypatch.setattr(updater_mod, "_latest_pypi_version", lambda package: "1.0.0") + + updater = AutoUpdater() + assert updater._check_and_update() is True