mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-29 18:46:59 +00:00
test(runtime): cover managed SQLite cutover (E-949)
This commit is contained in:
Justin Bennington
Teknium
parent
17bf3c8283
commit
bbee3011fd
4 changed files with 433 additions and 1 deletions
|
|
@ -359,10 +359,17 @@ class TestCmdUpdateBranchFallback:
|
|||
branch="main", verify_ok=True, commit_count="0"
|
||||
)
|
||||
|
||||
cmd_update(mock_args)
|
||||
with patch("hermes_cli.managed_uv.update_managed_uv") as mock_uv_update, \
|
||||
patch(
|
||||
"hermes_cli.managed_uv.ensure_uv",
|
||||
return_value=None,
|
||||
) as mock_uv_ensure:
|
||||
cmd_update(mock_args)
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "Already up to date!" in captured.out
|
||||
mock_uv_update.assert_called_once_with()
|
||||
mock_uv_ensure.assert_called_once_with()
|
||||
|
||||
# Should NOT have called pull
|
||||
commands = [" ".join(str(a) for a in c.args[0]) for c in mock_run.call_args_list]
|
||||
|
|
|
|||
|
|
@ -4,7 +4,9 @@ from __future__ import annotations
|
|||
|
||||
import os
|
||||
import stat
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
|
@ -21,6 +23,40 @@ def _make_executable(path: Path) -> None:
|
|||
path.chmod(path.stat().st_mode | stat.S_IEXEC)
|
||||
|
||||
|
||||
def _runtime_info(
|
||||
executable: Path,
|
||||
sqlite_version: tuple[int, int, int],
|
||||
):
|
||||
from hermes_cli.sqlite_runtime import SQLiteRuntimeInfo
|
||||
|
||||
return SQLiteRuntimeInfo(
|
||||
executable=executable,
|
||||
base_prefix=executable.parent.parent,
|
||||
python_version=(3, 11, 15),
|
||||
sqlite_version=sqlite_version,
|
||||
sqlite_version_string=".".join(str(part) for part in sqlite_version),
|
||||
sqlite_source_id=f"source-{sqlite_version}",
|
||||
)
|
||||
|
||||
|
||||
def _make_runtime_install(
|
||||
tmp_path: Path,
|
||||
*,
|
||||
windows: bool = False,
|
||||
) -> tuple[Path, Path, Path]:
|
||||
root = tmp_path / "checkout"
|
||||
root.mkdir()
|
||||
(root / "pyproject.toml").write_text("[project]\n", encoding="utf-8")
|
||||
live = root / "venv"
|
||||
bin_dir = live / ("Scripts" if windows else "bin")
|
||||
bin_dir.mkdir(parents=True)
|
||||
python = bin_dir / ("python.exe" if windows else "python")
|
||||
python.write_text("live interpreter", encoding="utf-8")
|
||||
sentinel = live / "sentinel"
|
||||
sentinel.write_text("live", encoding="utf-8")
|
||||
return root, live, sentinel
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# managed_uv_path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -229,6 +265,301 @@ class TestUpdateManagedUv:
|
|||
# Still returns the path — failure is non-fatal
|
||||
assert result == str(tmp_path / "bin" / "uv")
|
||||
|
||||
def test_old_updater_api_triggers_runtime_repair(self, tmp_path):
|
||||
"""The pre-pull main.py call site must activate the fresh module hook."""
|
||||
from hermes_cli.managed_uv import RuntimeRepairResult, update_managed_uv
|
||||
|
||||
uv = tmp_path / "bin" / "uv"
|
||||
_make_executable(uv)
|
||||
with patch("hermes_cli.managed_uv.get_hermes_home", return_value=tmp_path), \
|
||||
patch("hermes_cli.managed_uv.platform.system", return_value="Linux"), \
|
||||
patch("hermes_cli.managed_uv.subprocess.run") as mock_run, \
|
||||
patch(
|
||||
"hermes_cli.managed_uv.repair_vulnerable_runtime",
|
||||
return_value=RuntimeRepairResult(
|
||||
"repaired",
|
||||
sqlite_before="3.50.4",
|
||||
sqlite_after="3.53.1",
|
||||
),
|
||||
) as mock_repair:
|
||||
mock_run.side_effect = [
|
||||
MagicMock(returncode=0, stdout="", stderr=""),
|
||||
MagicMock(returncode=0, stdout="uv 0.11.31\n", stderr=""),
|
||||
]
|
||||
|
||||
result = update_managed_uv()
|
||||
|
||||
assert result == str(uv)
|
||||
mock_repair.assert_called_once_with(str(uv))
|
||||
|
||||
|
||||
class TestManagedPythonStore:
|
||||
def test_store_is_checkout_scoped_across_profiles(self, tmp_path, monkeypatch):
|
||||
from hermes_cli.managed_uv import managed_python_install_dir
|
||||
|
||||
checkout = tmp_path / "checkout"
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profiles" / "alpha"))
|
||||
alpha = managed_python_install_dir(checkout)
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profiles" / "beta"))
|
||||
beta = managed_python_install_dir(checkout)
|
||||
|
||||
expected = checkout / ".hermes-runtime" / "python"
|
||||
assert alpha == expected
|
||||
assert beta == expected
|
||||
|
||||
def test_environment_is_private_and_sanitized(self, tmp_path):
|
||||
from hermes_cli.managed_uv import managed_python_env
|
||||
|
||||
checkout = tmp_path / "checkout"
|
||||
base_env = {
|
||||
"KEEP_ME": "yes",
|
||||
"CONDA_DEFAULT_ENV": "poison",
|
||||
"CONDA_PREFIX": "/poison/conda",
|
||||
"UV_PROJECT_ENVIRONMENT": "/poison/project",
|
||||
"UV_NO_MANAGED_PYTHON": "1",
|
||||
"UV_PYTHON": "/poison/python",
|
||||
"UV_PYTHON_DOWNLOADS": "never",
|
||||
"UV_SYSTEM_PYTHON": "1",
|
||||
"VIRTUAL_ENV": "/poison/venv",
|
||||
"PYTHONHOME": "/poison/home",
|
||||
"PYTHONPATH": "/poison/path",
|
||||
}
|
||||
|
||||
env = managed_python_env(checkout, base_env=base_env)
|
||||
|
||||
assert env["KEEP_ME"] == "yes"
|
||||
assert env["UV_MANAGED_PYTHON"] == "1"
|
||||
assert env["UV_NO_CONFIG"] == "1"
|
||||
assert env["UV_PYTHON_INSTALL_BIN"] == "0"
|
||||
assert env["UV_PYTHON_INSTALL_REGISTRY"] == "0"
|
||||
assert env["UV_PYTHON_INSTALL_DIR"] == str(
|
||||
checkout / ".hermes-runtime" / "python"
|
||||
)
|
||||
for key in (
|
||||
"CONDA_DEFAULT_ENV",
|
||||
"CONDA_PREFIX",
|
||||
"UV_PROJECT_ENVIRONMENT",
|
||||
"UV_NO_MANAGED_PYTHON",
|
||||
"UV_PYTHON",
|
||||
"UV_PYTHON_DOWNLOADS",
|
||||
"UV_SYSTEM_PYTHON",
|
||||
"VIRTUAL_ENV",
|
||||
"PYTHONHOME",
|
||||
"PYTHONPATH",
|
||||
):
|
||||
assert key not in env
|
||||
assert base_env["PYTHONHOME"] == "/poison/home"
|
||||
|
||||
|
||||
class TestRuntimeRepair:
|
||||
def test_safe_runtime_is_a_noop(self, tmp_path):
|
||||
from hermes_cli.managed_uv import repair_vulnerable_runtime
|
||||
|
||||
root, live, sentinel = _make_runtime_install(tmp_path)
|
||||
current = _runtime_info(live / "bin" / "python", (3, 53, 1))
|
||||
with patch("hermes_cli.managed_uv.platform.system", return_value="Linux"), \
|
||||
patch(
|
||||
"hermes_cli.managed_uv.probe_sqlite_runtime",
|
||||
return_value=current,
|
||||
), \
|
||||
patch(
|
||||
"hermes_cli.managed_uv._install_safe_python_generation"
|
||||
) as mock_install:
|
||||
result = repair_vulnerable_runtime("uv", project_root=root)
|
||||
|
||||
assert result.status == "safe"
|
||||
assert result.sqlite_before == "3.53.1"
|
||||
assert result.sqlite_after == "3.53.1"
|
||||
assert sentinel.read_text(encoding="utf-8") == "live"
|
||||
assert not (root / ".hermes-runtime").exists()
|
||||
mock_install.assert_not_called()
|
||||
|
||||
def test_failed_candidate_preserves_live_venv(self, tmp_path):
|
||||
from hermes_cli.managed_uv import (
|
||||
_acquire_repair_lock,
|
||||
_release_repair_lock,
|
||||
repair_vulnerable_runtime,
|
||||
)
|
||||
|
||||
root, live, sentinel = _make_runtime_install(tmp_path)
|
||||
current = _runtime_info(live / "bin" / "python", (3, 50, 4))
|
||||
generation = root / ".hermes-runtime" / "python" / "generation-test"
|
||||
candidate_python = generation / "bin" / "python"
|
||||
candidate_python.parent.mkdir(parents=True)
|
||||
candidate_python.write_text("candidate interpreter", encoding="utf-8")
|
||||
fixed = _runtime_info(candidate_python, (3, 53, 1))
|
||||
|
||||
with patch("hermes_cli.managed_uv.platform.system", return_value="Linux"), \
|
||||
patch(
|
||||
"hermes_cli.managed_uv.probe_sqlite_runtime",
|
||||
side_effect=[current, current],
|
||||
), \
|
||||
patch(
|
||||
"hermes_cli.managed_uv._install_safe_python_generation",
|
||||
return_value=(generation, candidate_python, fixed),
|
||||
), \
|
||||
patch(
|
||||
"hermes_cli.managed_uv._stage_candidate_venv",
|
||||
return_value=None,
|
||||
):
|
||||
result = repair_vulnerable_runtime("uv", project_root=root)
|
||||
|
||||
assert result.status == "failed"
|
||||
assert "replacement environment" in result.detail
|
||||
assert sentinel.read_text(encoding="utf-8") == "live"
|
||||
assert (live / "bin" / "python").read_text(encoding="utf-8") == (
|
||||
"live interpreter"
|
||||
)
|
||||
assert not generation.exists()
|
||||
reacquired = _acquire_repair_lock(root / ".hermes-runtime")
|
||||
assert reacquired is not None
|
||||
_release_repair_lock(reacquired)
|
||||
|
||||
def test_windows_holders_refuse_runtime_mutation(self, tmp_path, monkeypatch):
|
||||
from hermes_cli.managed_uv import repair_vulnerable_runtime
|
||||
|
||||
root, live, sentinel = _make_runtime_install(tmp_path, windows=True)
|
||||
current = _runtime_info(live / "Scripts" / "python.exe", (3, 50, 4))
|
||||
old_main = SimpleNamespace(
|
||||
_detect_venv_python_processes=lambda: [
|
||||
(1729, "python.exe", "hermes gateway run")
|
||||
]
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "hermes_cli.main", old_main)
|
||||
|
||||
with patch("hermes_cli.managed_uv.platform.system", return_value="Windows"), \
|
||||
patch(
|
||||
"hermes_cli.managed_uv.probe_sqlite_runtime",
|
||||
return_value=current,
|
||||
), \
|
||||
patch(
|
||||
"hermes_cli.managed_uv._install_safe_python_generation"
|
||||
) as mock_install:
|
||||
result = repair_vulnerable_runtime("uv.exe", project_root=root)
|
||||
|
||||
assert result.status == "skipped"
|
||||
assert "PID 1729" in result.detail
|
||||
assert sentinel.read_text(encoding="utf-8") == "live"
|
||||
assert not (root / ".hermes-runtime").exists()
|
||||
mock_install.assert_not_called()
|
||||
|
||||
|
||||
class TestRuntimeCutover:
|
||||
def test_os_lock_blocks_concurrent_repair_and_releases(self, tmp_path):
|
||||
from hermes_cli.managed_uv import _acquire_repair_lock, _release_repair_lock
|
||||
|
||||
runtime_root = tmp_path / ".hermes-runtime"
|
||||
first = _acquire_repair_lock(runtime_root)
|
||||
assert first is not None
|
||||
assert _acquire_repair_lock(runtime_root) is None
|
||||
|
||||
_release_repair_lock(first)
|
||||
second = _acquire_repair_lock(runtime_root)
|
||||
assert second is not None
|
||||
_release_repair_lock(second)
|
||||
|
||||
def test_failed_smoke_with_empty_output_has_stable_detail(self, tmp_path):
|
||||
from hermes_cli.managed_uv import _smoke_candidate_venv
|
||||
|
||||
candidate = tmp_path / "venv"
|
||||
candidate.mkdir()
|
||||
fixed = _runtime_info(candidate / "bin" / "python", (3, 53, 1))
|
||||
failed = MagicMock(returncode=1, stdout=" \n", stderr="\n")
|
||||
with patch(
|
||||
"hermes_cli.managed_uv.probe_sqlite_runtime",
|
||||
return_value=fixed,
|
||||
), patch("hermes_cli.managed_uv.subprocess.run", return_value=failed):
|
||||
healthy, detail, info = _smoke_candidate_venv(candidate)
|
||||
|
||||
assert healthy is False
|
||||
assert detail == "core import smoke failed"
|
||||
assert info == fixed
|
||||
|
||||
def test_successfully_renames_candidate_into_live_path(self, tmp_path):
|
||||
from hermes_cli.managed_uv import _cut_over_candidate
|
||||
|
||||
root, _, _ = _make_runtime_install(tmp_path)
|
||||
runtime_root = root / ".hermes-runtime"
|
||||
candidate = runtime_root / "venv-candidate-test"
|
||||
candidate.mkdir(parents=True)
|
||||
(candidate / "sentinel").write_text("candidate", encoding="utf-8")
|
||||
fixed = _runtime_info(candidate / "bin" / "python", (3, 53, 1))
|
||||
|
||||
with patch(
|
||||
"hermes_cli.managed_uv._smoke_candidate_venv",
|
||||
return_value=(True, "", fixed),
|
||||
):
|
||||
ok, backup, info, detail = _cut_over_candidate(
|
||||
candidate,
|
||||
project_root=root,
|
||||
)
|
||||
|
||||
assert ok is True
|
||||
assert detail == ""
|
||||
assert info == fixed
|
||||
assert backup is not None
|
||||
assert (root / "venv" / "sentinel").read_text(encoding="utf-8") == (
|
||||
"candidate"
|
||||
)
|
||||
assert (backup / "sentinel").read_text(encoding="utf-8") == "live"
|
||||
assert not candidate.exists()
|
||||
|
||||
def test_post_swap_smoke_failure_rolls_back_live_venv(self, tmp_path):
|
||||
from hermes_cli.managed_uv import _cut_over_candidate
|
||||
|
||||
root, live, sentinel = _make_runtime_install(tmp_path)
|
||||
runtime_root = root / ".hermes-runtime"
|
||||
candidate = runtime_root / "venv-candidate-test"
|
||||
candidate.mkdir(parents=True)
|
||||
(candidate / "sentinel").write_text("candidate", encoding="utf-8")
|
||||
rejected_info = _runtime_info(candidate / "bin" / "python", (3, 50, 4))
|
||||
|
||||
with patch(
|
||||
"hermes_cli.managed_uv._smoke_candidate_venv",
|
||||
return_value=(False, "core import smoke failed", rejected_info),
|
||||
):
|
||||
ok, backup, info, detail = _cut_over_candidate(
|
||||
candidate,
|
||||
project_root=root,
|
||||
)
|
||||
|
||||
assert ok is False
|
||||
assert backup is None
|
||||
assert info == rejected_info
|
||||
assert "post-cutover smoke failed" in detail
|
||||
assert sentinel.read_text(encoding="utf-8") == "live"
|
||||
assert (live / "bin" / "python").read_text(encoding="utf-8") == (
|
||||
"live interpreter"
|
||||
)
|
||||
assert not candidate.exists()
|
||||
assert not list(runtime_root.glob("venv-rejected-*"))
|
||||
|
||||
def test_smoke_exception_after_swap_rolls_back_live_venv(self, tmp_path):
|
||||
from hermes_cli.managed_uv import _cut_over_candidate
|
||||
|
||||
root, live, sentinel = _make_runtime_install(tmp_path)
|
||||
candidate = root / ".hermes-runtime" / "venv-candidate-test"
|
||||
candidate.mkdir(parents=True)
|
||||
(candidate / "sentinel").write_text("candidate", encoding="utf-8")
|
||||
|
||||
with patch(
|
||||
"hermes_cli.managed_uv._smoke_candidate_venv",
|
||||
side_effect=RuntimeError("probe crashed"),
|
||||
):
|
||||
ok, backup, info, detail = _cut_over_candidate(
|
||||
candidate,
|
||||
project_root=root,
|
||||
)
|
||||
|
||||
assert ok is False
|
||||
assert backup is None
|
||||
assert info is None
|
||||
assert "probe crashed" in detail
|
||||
assert sentinel.read_text(encoding="utf-8") == "live"
|
||||
assert (live / "bin" / "python").read_text(encoding="utf-8") == (
|
||||
"live interpreter"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _install_uv internals
|
||||
|
|
|
|||
93
tests/hermes_cli/test_sqlite_runtime.py
Normal file
93
tests/hermes_cli/test_sqlite_runtime.py
Normal file
|
|
@ -0,0 +1,93 @@
|
|||
"""Behavioral tests for exact-interpreter SQLite runtime inspection."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shlex
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.sqlite_runtime import (
|
||||
is_sqlite_wal_reset_vulnerable,
|
||||
probe_sqlite_runtime,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("version", "expected"),
|
||||
[
|
||||
((3, 6, 23), False),
|
||||
((3, 7, 0), True),
|
||||
((3, 44, 5), True),
|
||||
((3, 44, 6), False),
|
||||
((3, 45, 0), True),
|
||||
((3, 50, 6), True),
|
||||
((3, 50, 7), False),
|
||||
((3, 51, 2), True),
|
||||
((3, 51, 3), False),
|
||||
((3, 53, 1), False),
|
||||
],
|
||||
)
|
||||
def test_wal_reset_vulnerability_matrix(
|
||||
version: tuple[int, ...],
|
||||
expected: bool,
|
||||
) -> None:
|
||||
assert is_sqlite_wal_reset_vulnerable(version) is expected
|
||||
|
||||
|
||||
def test_probe_reports_the_requested_interpreters_linked_sqlite() -> None:
|
||||
info = probe_sqlite_runtime(sys.executable)
|
||||
|
||||
assert info is not None
|
||||
assert info.executable.resolve() == Path(sys.executable).resolve()
|
||||
assert info.base_prefix.resolve() == Path(sys.base_prefix).resolve()
|
||||
assert info.python_version == sys.version_info[:3]
|
||||
assert info.sqlite_version == sqlite3.sqlite_version_info
|
||||
assert info.sqlite_version_string == sqlite3.sqlite_version
|
||||
|
||||
with sqlite3.connect(":memory:") as conn:
|
||||
source_id = conn.execute("SELECT sqlite_source_id()").fetchone()[0]
|
||||
assert info.sqlite_source_id == source_id
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="uses a POSIX executable probe stub")
|
||||
def test_probe_uses_child_payload_and_sanitizes_python_environment(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
fake_python = tmp_path / "reported-python"
|
||||
payload = {
|
||||
"base_prefix": str(tmp_path / "reported-base"),
|
||||
"executable": str(fake_python),
|
||||
"python_version": [3, 11, 15],
|
||||
"sqlite_version": [9, 8, 7],
|
||||
"sqlite_version_string": "9.8.7-child",
|
||||
"sqlite_source_id": "child-source-id",
|
||||
}
|
||||
fake_python.write_text(
|
||||
"\n".join([
|
||||
"#!/bin/sh",
|
||||
'[ "$1" = "-I" ] && [ "$2" = "-c" ] || exit 10',
|
||||
'[ -z "${PYTHONHOME+x}" ] || exit 11',
|
||||
'[ -z "${PYTHONPATH+x}" ] || exit 12',
|
||||
f"printf '%s\\n' {shlex.quote(json.dumps(payload))}",
|
||||
])
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
fake_python.chmod(0o755)
|
||||
monkeypatch.setenv("PYTHONHOME", str(tmp_path / "poison-home"))
|
||||
monkeypatch.setenv("PYTHONPATH", str(tmp_path / "poison-path"))
|
||||
|
||||
info = probe_sqlite_runtime(fake_python)
|
||||
|
||||
assert info is not None
|
||||
assert info.executable == fake_python
|
||||
assert info.base_prefix == tmp_path / "reported-base"
|
||||
assert info.sqlite_version == (9, 8, 7)
|
||||
assert info.sqlite_version_string == "9.8.7-child"
|
||||
assert info.sqlite_source_id == "child-source-id"
|
||||
|
|
@ -186,5 +186,6 @@ def test_doctor_warns_without_adding_issues(monkeypatch, tmp_path, capsys):
|
|||
assert "SQLite" in out
|
||||
assert "3.50.4" in out
|
||||
assert "WAL-reset" in out
|
||||
assert "hermes update" in out
|
||||
# No longer appended to the blocking issues summary.
|
||||
assert "Linked SQLite is vulnerable" not in out
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue