248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""Tests for plugin auto-update: update --all sweep, autoupdate flag, startup sweep.
|
|
|
|
Inspired by Copilot CLI v1.0.79's marketplace ``autoUpdate`` setting.
|
|
Real-git E2E where the behavior depends on git (pull, revision recording),
|
|
mocks only for pure dispatch/throttle logic.
|
|
"""
|
|
|
|
import json
|
|
import subprocess as sp
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import hermes_cli.plugins_cmd as pc
|
|
|
|
|
|
def _git(cwd, *args):
|
|
r = sp.run(["git", *args], cwd=str(cwd), capture_output=True, text=True)
|
|
assert r.returncode == 0, r.stderr
|
|
return r.stdout
|
|
|
|
|
|
def _make_plugin_env(tmp_path, monkeypatch, names=("alpha",)):
|
|
"""Create a fake HERMES_HOME with git-installed plugins cloned from origins."""
|
|
home = tmp_path / "hermes-home"
|
|
plugins_dir = home / "plugins"
|
|
plugins_dir.mkdir(parents=True)
|
|
monkeypatch.setattr(pc, "get_hermes_home", lambda: home)
|
|
monkeypatch.setattr(pc, "_plugins_dir", lambda: plugins_dir)
|
|
|
|
origins = {}
|
|
for name in names:
|
|
origin = tmp_path / f"origin-{name}"
|
|
origin.mkdir()
|
|
_git(origin, "init", "-q", "-b", "main")
|
|
_git(origin, "config", "user.email", "t@t")
|
|
_git(origin, "config", "user.name", "t")
|
|
(origin / "plugin.yaml").write_text(f"name: {name}\n", encoding="utf-8")
|
|
(origin / "mod.py").write_text("VALUE = 1\n", encoding="utf-8")
|
|
_git(origin, "add", ".")
|
|
_git(origin, "commit", "-qm", "init")
|
|
checkout = plugins_dir / name
|
|
_git(tmp_path, "clone", "-q", str(origin), str(checkout))
|
|
_git(checkout, "config", "user.email", "t@t")
|
|
_git(checkout, "config", "user.name", "t")
|
|
origins[name] = origin
|
|
return home, plugins_dir, origins
|
|
|
|
|
|
def _advance_origin(origin, value):
|
|
(origin / "mod.py").write_text(f"VALUE = {value}\n", encoding="utf-8")
|
|
_git(origin, "add", ".")
|
|
_git(origin, "commit", "-qm", f"bump {value}")
|
|
|
|
|
|
def _write_metadata(home, metadata):
|
|
path = home / "plugins" / pc._INSTALL_METADATA_FILE
|
|
path.write_text(json.dumps(metadata), encoding="utf-8")
|
|
|
|
|
|
def _read_metadata(home):
|
|
path = home / "plugins" / pc._INSTALL_METADATA_FILE
|
|
return json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
|
|
|
|
|
|
class TestIterUpdatablePlugins:
|
|
def test_lists_git_dirs_only(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, _ = _make_plugin_env(tmp_path, monkeypatch, ("alpha", "beta"))
|
|
(plugins_dir / "not-git").mkdir() # plain dir — excluded
|
|
(plugins_dir / ".hidden").mkdir() # dot dir — excluded
|
|
names = [t.name for t, _ in pc._iter_updatable_plugins()]
|
|
assert names == ["alpha", "beta"]
|
|
|
|
def test_missing_plugins_dir_returns_empty(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(pc, "_plugins_dir", lambda: tmp_path / "nope")
|
|
assert pc._iter_updatable_plugins() == []
|
|
|
|
|
|
class TestUpdateOnePluginDir:
|
|
def test_pull_updates_and_records_revision(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, origins = _make_plugin_env(tmp_path, monkeypatch)
|
|
_write_metadata(home, {"alpha": {"pinned": False, "revision": "old", "source": "x"}})
|
|
_advance_origin(origins["alpha"], 2)
|
|
|
|
res = pc._update_one_plugin_dir(plugins_dir / "alpha", {})
|
|
assert res["ok"] is True and res["unchanged"] is False
|
|
assert (plugins_dir / "alpha" / "mod.py").read_text() == "VALUE = 2\n"
|
|
rev = _read_metadata(home)["alpha"]["revision"]
|
|
assert len(rev) == 40 and rev != "old"
|
|
|
|
def test_unchanged_pull(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, _ = _make_plugin_env(tmp_path, monkeypatch)
|
|
res = pc._update_one_plugin_dir(plugins_dir / "alpha", {})
|
|
assert res["ok"] is True and res["unchanged"] is True
|
|
|
|
def test_pinned_is_skipped(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, origins = _make_plugin_env(tmp_path, monkeypatch)
|
|
_advance_origin(origins["alpha"], 3)
|
|
res = pc._update_one_plugin_dir(
|
|
plugins_dir / "alpha", {"pinned": True, "revision": "a" * 40}
|
|
)
|
|
assert res["skipped"] is True
|
|
# Checkout untouched
|
|
assert (plugins_dir / "alpha" / "mod.py").read_text() == "VALUE = 1\n"
|
|
|
|
def test_clears_stale_bytecode(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, origins = _make_plugin_env(tmp_path, monkeypatch)
|
|
cache = plugins_dir / "alpha" / "__pycache__"
|
|
cache.mkdir()
|
|
(cache / "mod.cpython-311.pyc").write_bytes(b"stale")
|
|
_advance_origin(origins["alpha"], 4)
|
|
res = pc._update_one_plugin_dir(plugins_dir / "alpha", {})
|
|
assert res["ok"] is True and res["unchanged"] is False
|
|
assert not cache.exists()
|
|
|
|
|
|
class TestUpdateAllPlugins:
|
|
def test_mixed_sweep(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, origins = _make_plugin_env(
|
|
tmp_path, monkeypatch, ("alpha", "beta", "gamma")
|
|
)
|
|
_write_metadata(home, {"beta": {"pinned": True, "revision": "b" * 40, "source": "x"}})
|
|
_advance_origin(origins["alpha"], 9)
|
|
|
|
console = MagicMock()
|
|
results = {r["name"]: r for r in pc._update_all_plugins(console)}
|
|
assert results["alpha"]["unchanged"] is False
|
|
assert results["beta"]["skipped"] is True
|
|
assert results["gamma"]["unchanged"] is True
|
|
|
|
|
|
class TestCmdUpdateDispatch:
|
|
def test_all_flag_runs_sweep(self, tmp_path, monkeypatch):
|
|
with patch.object(pc, "_update_all_plugins", return_value=[]) as sweep:
|
|
pc.cmd_update(None, all_plugins=True)
|
|
sweep.assert_called_once()
|
|
|
|
def test_name_plus_all_rejected(self):
|
|
with pytest.raises(SystemExit):
|
|
pc.cmd_update("alpha", all_plugins=True)
|
|
|
|
def test_no_name_no_all_rejected(self):
|
|
with pytest.raises(SystemExit):
|
|
pc.cmd_update(None)
|
|
|
|
|
|
class TestCmdAutoupdate:
|
|
def test_enable_persists_flag(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, _ = _make_plugin_env(tmp_path, monkeypatch)
|
|
_write_metadata(home, {"alpha": {"pinned": False, "revision": "r", "source": "s"}})
|
|
pc.cmd_autoupdate("alpha", "on")
|
|
assert _read_metadata(home)["alpha"]["auto_update"] is True
|
|
|
|
def test_disable_removes_flag(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, _ = _make_plugin_env(tmp_path, monkeypatch)
|
|
_write_metadata(
|
|
home,
|
|
{"alpha": {"pinned": False, "revision": "r", "source": "s", "auto_update": True}},
|
|
)
|
|
pc.cmd_autoupdate("alpha", "off")
|
|
assert "auto_update" not in _read_metadata(home)["alpha"]
|
|
|
|
def test_pinned_plugin_rejected(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, _ = _make_plugin_env(tmp_path, monkeypatch)
|
|
_write_metadata(home, {"alpha": {"pinned": True, "revision": "a" * 40, "source": "s"}})
|
|
with pytest.raises(SystemExit):
|
|
pc.cmd_autoupdate("alpha", "on")
|
|
|
|
def test_non_git_plugin_rejected(self, tmp_path, monkeypatch):
|
|
home = tmp_path / "hh"
|
|
plugins_dir = home / "plugins"
|
|
(plugins_dir / "plain").mkdir(parents=True)
|
|
monkeypatch.setattr(pc, "get_hermes_home", lambda: home)
|
|
monkeypatch.setattr(pc, "_plugins_dir", lambda: plugins_dir)
|
|
with pytest.raises(SystemExit):
|
|
pc.cmd_autoupdate("plain", "on")
|
|
|
|
def test_missing_plugin_rejected(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, _ = _make_plugin_env(tmp_path, monkeypatch)
|
|
with pytest.raises(SystemExit):
|
|
pc.cmd_autoupdate("ghost", "on")
|
|
|
|
|
|
class TestStartupSweep:
|
|
def test_only_flagged_plugins_pulled(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, origins = _make_plugin_env(
|
|
tmp_path, monkeypatch, ("alpha", "beta")
|
|
)
|
|
_write_metadata(
|
|
home,
|
|
{
|
|
"alpha": {"pinned": False, "revision": "r", "source": "s", "auto_update": True},
|
|
"beta": {"pinned": False, "revision": "r", "source": "s"},
|
|
},
|
|
)
|
|
_advance_origin(origins["alpha"], 7)
|
|
_advance_origin(origins["beta"], 7)
|
|
|
|
results = pc.run_startup_auto_update_sweep(force=True)
|
|
assert [r["name"] for r in results] == ["alpha"]
|
|
assert (plugins_dir / "alpha" / "mod.py").read_text() == "VALUE = 7\n"
|
|
assert (plugins_dir / "beta" / "mod.py").read_text() == "VALUE = 1\n"
|
|
|
|
def test_no_flagged_plugins_is_noop(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, _ = _make_plugin_env(tmp_path, monkeypatch)
|
|
assert pc.run_startup_auto_update_sweep(force=True) == []
|
|
# No stamp written when nothing is opted in
|
|
assert not pc._autoupdate_stamp_path().exists()
|
|
|
|
def test_throttled_by_stamp(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, origins = _make_plugin_env(tmp_path, monkeypatch)
|
|
_write_metadata(
|
|
home,
|
|
{"alpha": {"pinned": False, "revision": "r", "source": "s", "auto_update": True}},
|
|
)
|
|
stamp = pc._autoupdate_stamp_path()
|
|
stamp.parent.mkdir(parents=True, exist_ok=True)
|
|
stamp.touch() # fresh stamp → throttled
|
|
assert pc.run_startup_auto_update_sweep() == []
|
|
|
|
def test_stale_stamp_runs(self, tmp_path, monkeypatch):
|
|
import os
|
|
|
|
home, plugins_dir, origins = _make_plugin_env(tmp_path, monkeypatch)
|
|
_write_metadata(
|
|
home,
|
|
{"alpha": {"pinned": False, "revision": "r", "source": "s", "auto_update": True}},
|
|
)
|
|
stamp = pc._autoupdate_stamp_path()
|
|
stamp.parent.mkdir(parents=True, exist_ok=True)
|
|
stamp.touch()
|
|
old = time.time() - (pc._AUTOUPDATE_INTERVAL_SECONDS + 60)
|
|
os.utime(stamp, (old, old))
|
|
results = pc.run_startup_auto_update_sweep()
|
|
assert [r["name"] for r in results] == ["alpha"]
|
|
# Stamp refreshed
|
|
assert time.time() - stamp.stat().st_mtime < 60
|
|
|
|
def test_never_raises_on_broken_checkout(self, tmp_path, monkeypatch):
|
|
home, plugins_dir, _ = _make_plugin_env(tmp_path, monkeypatch)
|
|
_write_metadata(
|
|
home,
|
|
{"alpha": {"pinned": False, "revision": "r", "source": "s", "auto_update": True}},
|
|
)
|
|
with patch.object(pc, "_update_one_plugin_dir", side_effect=RuntimeError("boom")):
|
|
assert pc.run_startup_auto_update_sweep(force=True) == []
|