fix(install): scrap rebuild venv
This commit is contained in:
@ -59,19 +59,14 @@ def _patch_managed_uv(request):
|
||||
return shutil.which("uv")
|
||||
|
||||
def _fake_ensure_uv():
|
||||
path = shutil.which("uv")
|
||||
return (path, False) # never freshly bootstrapped in tests
|
||||
return shutil.which("uv")
|
||||
|
||||
def _fake_update_managed_uv():
|
||||
return None # never actually self-update in tests
|
||||
|
||||
def _fake_rebuild_venv(*args, **kwargs):
|
||||
return True # no-op in tests
|
||||
|
||||
with patch("hermes_cli.managed_uv.resolve_uv", side_effect=_fake_resolve_uv), \
|
||||
patch("hermes_cli.managed_uv.ensure_uv", side_effect=_fake_ensure_uv), \
|
||||
patch("hermes_cli.managed_uv.update_managed_uv", side_effect=_fake_update_managed_uv), \
|
||||
patch("hermes_cli.managed_uv.rebuild_venv", side_effect=_fake_rebuild_venv):
|
||||
patch("hermes_cli.managed_uv.update_managed_uv", side_effect=_fake_update_managed_uv):
|
||||
yield
|
||||
|
||||
|
||||
|
||||
@ -76,11 +76,10 @@ class TestEnsureUv:
|
||||
_make_executable(tmp_path / "bin" / "uv")
|
||||
with patch("hermes_cli.managed_uv.get_hermes_home", return_value=tmp_path):
|
||||
from hermes_cli.managed_uv import ensure_uv
|
||||
path, fresh = ensure_uv()
|
||||
path = ensure_uv()
|
||||
assert path == str(tmp_path / "bin" / "uv")
|
||||
assert fresh is False
|
||||
|
||||
def test_installs_if_missing_sets_bootstrap_flag(self, tmp_path):
|
||||
def test_installs_if_missing(self, tmp_path):
|
||||
with patch("hermes_cli.managed_uv.get_hermes_home", return_value=tmp_path), \
|
||||
patch("hermes_cli.managed_uv._install_uv") as mock_install:
|
||||
# Simulate the installer creating the binary
|
||||
@ -89,127 +88,16 @@ class TestEnsureUv:
|
||||
mock_install.side_effect = fake_install
|
||||
|
||||
from hermes_cli.managed_uv import ensure_uv
|
||||
path, fresh = ensure_uv()
|
||||
path = ensure_uv()
|
||||
assert path == str(tmp_path / "bin" / "uv")
|
||||
assert fresh is True
|
||||
mock_install.assert_called_once()
|
||||
|
||||
def test_install_failure_returns_none_false(self, tmp_path):
|
||||
def test_install_failure_returns_none(self, tmp_path):
|
||||
with patch("hermes_cli.managed_uv.get_hermes_home", return_value=tmp_path), \
|
||||
patch("hermes_cli.managed_uv._install_uv", side_effect=RuntimeError("network down")):
|
||||
from hermes_cli.managed_uv import ensure_uv
|
||||
path, fresh = ensure_uv()
|
||||
path = ensure_uv()
|
||||
assert path is None
|
||||
assert fresh is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# rebuild_venv
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRebuildVenv:
|
||||
def test_moves_old_venv_aside_and_creates_new(self, tmp_path):
|
||||
"""The old venv is moved aside to <venv>.old (never rmtree'd in place),
|
||||
uv is invoked with --clear, the moved-aside backup is removed on
|
||||
success, and the rebuilt interpreter is reported."""
|
||||
venv_dir = tmp_path / "venv"
|
||||
venv_dir.mkdir()
|
||||
(venv_dir / "old_file").write_text("stale")
|
||||
|
||||
uv_bin = str(tmp_path / "bin" / "uv")
|
||||
call_log: list[list[str]] = []
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
call_log.append(list(cmd))
|
||||
m = MagicMock(returncode=0, stderr="", stdout="")
|
||||
if len(cmd) >= 2 and cmd[1] == "venv":
|
||||
# Simulate uv creating the venv dir with a python interpreter
|
||||
bin_dir = venv_dir / ("Scripts" if os.name == "nt" else "bin")
|
||||
bin_dir.mkdir(parents=True, exist_ok=True)
|
||||
python_name = "python.exe" if os.name == "nt" else "python"
|
||||
(bin_dir / python_name).write_text("#!/bin/sh\necho Python 3.11.0")
|
||||
elif "--version" in cmd:
|
||||
m.stdout = "Python 3.11.0"
|
||||
return m
|
||||
|
||||
with patch("hermes_cli.managed_uv.subprocess.run", side_effect=fake_run):
|
||||
from hermes_cli.managed_uv import rebuild_venv
|
||||
result = rebuild_venv(uv_bin, venv_dir)
|
||||
|
||||
assert result is True
|
||||
# uv venv was invoked exactly once, always with --clear.
|
||||
venv_calls = [c for c in call_log if len(c) >= 2 and c[1] == "venv"]
|
||||
assert len(venv_calls) == 1, f"expected 1 venv call, got {venv_calls}"
|
||||
assert "--clear" in venv_calls[0]
|
||||
# The moved-aside backup is cleaned up after a successful rebuild.
|
||||
assert not (tmp_path / "venv.old").exists()
|
||||
|
||||
def test_aborts_without_deleting_when_venv_in_use(self, tmp_path):
|
||||
"""If os.replace fails (Windows file lock — venv in use), we must abort
|
||||
cleanly WITHOUT deleting the venv and WITHOUT invoking uv."""
|
||||
venv_dir = tmp_path / "venv"
|
||||
venv_dir.mkdir()
|
||||
(venv_dir / "locked") .write_text("held open")
|
||||
uv_bin = str(tmp_path / "bin" / "uv")
|
||||
call_log: list[list[str]] = []
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
call_log.append(list(cmd))
|
||||
return MagicMock(returncode=0, stderr="", stdout="")
|
||||
|
||||
with patch("hermes_cli.managed_uv.subprocess.run", side_effect=fake_run), \
|
||||
patch("hermes_cli.managed_uv.os.replace", side_effect=OSError("in use")):
|
||||
from hermes_cli.managed_uv import rebuild_venv
|
||||
result = rebuild_venv(uv_bin, venv_dir)
|
||||
|
||||
assert result is False
|
||||
# venv left fully intact, uv never invoked.
|
||||
assert venv_dir.exists() and (venv_dir / "locked").exists()
|
||||
assert [c for c in call_log if len(c) >= 2 and c[1] == "venv"] == []
|
||||
|
||||
def test_restores_backup_when_rebuild_fails(self, tmp_path):
|
||||
"""If uv venv exits non-zero, the moved-aside venv is restored so we
|
||||
never leave Hermes with no venv at all."""
|
||||
venv_dir = tmp_path / "venv"
|
||||
venv_dir.mkdir()
|
||||
(venv_dir / "marker").write_text("original")
|
||||
uv_bin = str(tmp_path / "bin" / "uv")
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
return MagicMock(returncode=1, stderr="boom", stdout="")
|
||||
|
||||
with patch("hermes_cli.managed_uv.subprocess.run", side_effect=fake_run):
|
||||
from hermes_cli.managed_uv import rebuild_venv
|
||||
result = rebuild_venv(uv_bin, venv_dir)
|
||||
|
||||
assert result is False
|
||||
# Original venv restored from the .old backup.
|
||||
assert venv_dir.exists() and (venv_dir / "marker").read_text() == "original"
|
||||
assert not (tmp_path / "venv.old").exists()
|
||||
|
||||
def test_rebuild_failure_returns_false(self, tmp_path):
|
||||
venv_dir = tmp_path / "venv"
|
||||
uv_bin = str(tmp_path / "bin" / "uv")
|
||||
|
||||
with patch("hermes_cli.managed_uv.subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=1, stderr="nope")
|
||||
from hermes_cli.managed_uv import rebuild_venv
|
||||
result = rebuild_venv(uv_bin, venv_dir)
|
||||
assert result is False
|
||||
|
||||
def test_rebuild_success_without_python_returns_false(self, tmp_path):
|
||||
"""uv can exit 0 yet leave no interpreter; that must not count as success
|
||||
(guard adapted from #38511)."""
|
||||
venv_dir = tmp_path / "venv"
|
||||
uv_bin = str(tmp_path / "bin" / "uv")
|
||||
|
||||
with patch("hermes_cli.managed_uv.subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
||||
from hermes_cli.managed_uv import rebuild_venv
|
||||
result = rebuild_venv(uv_bin, venv_dir)
|
||||
assert result is False
|
||||
# Returned before the `python --version` probe ran (only the uv venv call).
|
||||
assert mock_run.call_count == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@ -29,19 +29,14 @@ def _patch_managed_uv(request):
|
||||
return shutil.which("uv")
|
||||
|
||||
def _fake_ensure_uv():
|
||||
path = shutil.which("uv")
|
||||
return (path, False) # never freshly bootstrapped in tests
|
||||
return shutil.which("uv")
|
||||
|
||||
def _fake_update_managed_uv():
|
||||
return None # never actually self-update in tests
|
||||
|
||||
def _fake_rebuild_venv(*args, **kwargs):
|
||||
return True # no-op in tests
|
||||
|
||||
with patch("hermes_cli.managed_uv.resolve_uv", side_effect=_fake_resolve_uv), \
|
||||
patch("hermes_cli.managed_uv.ensure_uv", side_effect=_fake_ensure_uv), \
|
||||
patch("hermes_cli.managed_uv.update_managed_uv", side_effect=_fake_update_managed_uv), \
|
||||
patch("hermes_cli.managed_uv.rebuild_venv", side_effect=_fake_rebuild_venv):
|
||||
patch("hermes_cli.managed_uv.update_managed_uv", side_effect=_fake_update_managed_uv):
|
||||
yield
|
||||
|
||||
def test_stash_local_changes_if_needed_returns_none_when_tree_clean(monkeypatch, tmp_path):
|
||||
@ -423,41 +418,6 @@ def test_cmd_update_succeeds_with_extras(monkeypatch, tmp_path):
|
||||
assert ".[all]" in install_cmds[0]
|
||||
|
||||
|
||||
def test_cmd_update_aborts_when_fresh_managed_uv_rebuild_fails(monkeypatch, tmp_path):
|
||||
"""A failed fresh managed-uv venv rebuild must not continue into pip install
|
||||
(guard adapted from #38511)."""
|
||||
_setup_update_mocks(monkeypatch, tmp_path)
|
||||
monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/uv" if name == "uv" else None)
|
||||
monkeypatch.setattr(hermes_main, "_is_termux_env", lambda env=None: False)
|
||||
|
||||
recorded = []
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
recorded.append(cmd)
|
||||
# Tolerant matching: the update flow's exact git invocations vary by
|
||||
# checkout, so key off the verb. Branch detection must return a real name
|
||||
# and rev-list a parseable count, or the flow aborts early before it ever
|
||||
# reaches the venv rebuild this test exercises.
|
||||
if isinstance(cmd, (list, tuple)) and cmd and cmd[0] == "git":
|
||||
if "rev-parse" in cmd:
|
||||
return SimpleNamespace(stdout="main\n", stderr="", returncode=0)
|
||||
if "rev-list" in cmd:
|
||||
return SimpleNamespace(stdout="1\n", stderr="", returncode=0)
|
||||
if "pull" in cmd:
|
||||
return SimpleNamespace(stdout="Updating\n", stderr="", returncode=0)
|
||||
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
||||
|
||||
monkeypatch.setattr(hermes_main.subprocess, "run", fake_run)
|
||||
|
||||
with patch("hermes_cli.managed_uv.ensure_uv", return_value=("/usr/bin/uv", True)), \
|
||||
patch("hermes_cli.managed_uv.rebuild_venv", return_value=False), \
|
||||
pytest.raises(RuntimeError, match="venv rebuild failed"):
|
||||
hermes_main.cmd_update(SimpleNamespace())
|
||||
|
||||
install_cmds = [c for c in recorded if "pip" in c and "install" in c]
|
||||
assert install_cmds == []
|
||||
|
||||
|
||||
def test_install_with_optional_fallback_honors_custom_group(monkeypatch):
|
||||
"""Termux update path should target .[termux-all] when requested."""
|
||||
calls = []
|
||||
|
||||
@ -1,83 +0,0 @@
|
||||
"""Tests for ``_wait_for_interpreter_venv_ready`` in ``hermes_cli/main.py``.
|
||||
|
||||
During ``hermes update`` the managed-uv path can rebuild the project venv
|
||||
(rmtree + ``uv venv``) before the desktop-rebuild and profile-skills-sync
|
||||
steps spawn ``sys.executable``. If those children fire while the venv is
|
||||
mid-rewrite, the interpreter launcher aborts with ``No pyvenv.cfg file`` and
|
||||
the step spuriously "fails" on an otherwise-successful update. The helper
|
||||
waits for the marker to settle first.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from hermes_cli.main import _wait_for_interpreter_venv_ready
|
||||
|
||||
|
||||
def _make_fake_venv(tmp_path: Path, *, with_cfg: bool) -> Path:
|
||||
"""Create a venv-shaped dir and return the interpreter path inside it."""
|
||||
bin_name = "Scripts" if os.name == "nt" else "bin"
|
||||
bin_dir = tmp_path / bin_name
|
||||
bin_dir.mkdir(parents=True)
|
||||
py = bin_dir / ("python.exe" if os.name == "nt" else "python")
|
||||
py.write_text("#!/bin/sh\n")
|
||||
if with_cfg:
|
||||
(tmp_path / "pyvenv.cfg").write_text("home = /usr\n")
|
||||
return py
|
||||
|
||||
|
||||
class TestWaitForInterpreterVenvReady:
|
||||
def test_intact_venv_returns_immediately(self, tmp_path, monkeypatch):
|
||||
py = _make_fake_venv(tmp_path, with_cfg=True)
|
||||
monkeypatch.setattr("sys.executable", str(py))
|
||||
t0 = time.monotonic()
|
||||
assert _wait_for_interpreter_venv_ready(timeout=5) is True
|
||||
assert time.monotonic() - t0 < 0.5
|
||||
|
||||
def test_non_venv_interpreter_returns_immediately(self, tmp_path, monkeypatch):
|
||||
# A bare interpreter whose parent.parent has no bin/Scripts marker
|
||||
# dir is not venv-hosted; pyvenv.cfg is irrelevant.
|
||||
sys_py = tmp_path / "usr" / "bin" / "python"
|
||||
sys_py.parent.mkdir(parents=True)
|
||||
sys_py.write_text("#!/bin/sh\n")
|
||||
# Ensure parent.parent (tmp_path/usr) has no bin sibling shaped like a venv
|
||||
monkeypatch.setattr("sys.executable", str(sys_py))
|
||||
# parent.parent == tmp_path/usr; its "bin" child IS tmp_path/usr/bin
|
||||
# which exists — so this would look venv-ish. Use a deeper layout
|
||||
# where parent.parent has no bin marker:
|
||||
deep = tmp_path / "opt" / "py3" / "real" / "python"
|
||||
deep.parent.mkdir(parents=True)
|
||||
deep.write_text("#!/bin/sh\n")
|
||||
monkeypatch.setattr("sys.executable", str(deep))
|
||||
t0 = time.monotonic()
|
||||
assert _wait_for_interpreter_venv_ready(timeout=5) is True
|
||||
assert time.monotonic() - t0 < 0.5
|
||||
|
||||
def test_waits_for_cfg_to_appear(self, tmp_path, monkeypatch):
|
||||
py = _make_fake_venv(tmp_path, with_cfg=False)
|
||||
monkeypatch.setattr("sys.executable", str(py))
|
||||
|
||||
def _write_cfg_later():
|
||||
time.sleep(0.6)
|
||||
(tmp_path / "pyvenv.cfg").write_text("home = /usr\n")
|
||||
|
||||
th = threading.Thread(target=_write_cfg_later)
|
||||
th.start()
|
||||
try:
|
||||
t0 = time.monotonic()
|
||||
assert _wait_for_interpreter_venv_ready(timeout=5) is True
|
||||
elapsed = time.monotonic() - t0
|
||||
finally:
|
||||
th.join()
|
||||
assert 0.5 < elapsed < 2.0
|
||||
|
||||
def test_returns_false_when_cfg_never_appears(self, tmp_path, monkeypatch):
|
||||
py = _make_fake_venv(tmp_path, with_cfg=False)
|
||||
monkeypatch.setattr("sys.executable", str(py))
|
||||
t0 = time.monotonic()
|
||||
assert _wait_for_interpreter_venv_ready(timeout=1) is False
|
||||
assert 0.9 < time.monotonic() - t0 < 1.6
|
||||
@ -40,19 +40,14 @@ def _patch_managed_uv(request):
|
||||
return shutil.which("uv")
|
||||
|
||||
def _fake_ensure_uv():
|
||||
path = shutil.which("uv")
|
||||
return (path, False) # never freshly bootstrapped in tests
|
||||
return shutil.which("uv")
|
||||
|
||||
def _fake_update_managed_uv():
|
||||
return None # never actually self-update in tests
|
||||
|
||||
def _fake_rebuild_venv(*args, **kwargs):
|
||||
return True # no-op in tests
|
||||
|
||||
with patch("hermes_cli.managed_uv.resolve_uv", side_effect=_fake_resolve_uv), \
|
||||
patch("hermes_cli.managed_uv.ensure_uv", side_effect=_fake_ensure_uv), \
|
||||
patch("hermes_cli.managed_uv.update_managed_uv", side_effect=_fake_update_managed_uv), \
|
||||
patch("hermes_cli.managed_uv.rebuild_venv", side_effect=_fake_rebuild_venv):
|
||||
patch("hermes_cli.managed_uv.update_managed_uv", side_effect=_fake_update_managed_uv):
|
||||
yield
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user