test_disk_cleanup_plugin.py
1 """Tests for the disk-cleanup plugin. 2 3 Covers the bundled plugin at ``plugins/disk-cleanup/``: 4 5 * ``disk_cleanup`` library: track / forget / dry_run / quick / status, 6 ``is_safe_path`` and ``guess_category`` filtering. 7 * Plugin ``__init__``: ``post_tool_call`` hook auto-tracks files created 8 by ``write_file`` / ``terminal``; ``on_session_end`` hook runs quick 9 cleanup when anything was tracked during the turn. 10 * Slash command handler: status / dry-run / quick / track / forget / 11 unknown subcommand behaviours. 12 * Bundled-plugin discovery via ``PluginManager.discover_and_load``. 13 """ 14 15 import importlib 16 import json 17 import sys 18 from pathlib import Path 19 20 import pytest 21 22 23 @pytest.fixture(autouse=True) 24 def _isolate_env(tmp_path, monkeypatch): 25 """Isolate HERMES_HOME for each test. 26 27 The global hermetic fixture already redirects HERMES_HOME to a tempdir, 28 but we want the plugin to work with a predictable subpath. We reset 29 HERMES_HOME here for clarity. 30 """ 31 hermes_home = tmp_path / ".hermes" 32 hermes_home.mkdir() 33 monkeypatch.setenv("HERMES_HOME", str(hermes_home)) 34 yield hermes_home 35 36 37 def _load_lib(): 38 """Import the plugin's library module directly from the repo path.""" 39 repo_root = Path(__file__).resolve().parents[2] 40 lib_path = repo_root / "plugins" / "disk-cleanup" / "disk_cleanup.py" 41 spec = importlib.util.spec_from_file_location( 42 "disk_cleanup_under_test", lib_path 43 ) 44 mod = importlib.util.module_from_spec(spec) 45 spec.loader.exec_module(mod) 46 return mod 47 48 49 def _load_plugin_init(): 50 """Import the plugin's __init__.py (which depends on the library).""" 51 repo_root = Path(__file__).resolve().parents[2] 52 plugin_dir = repo_root / "plugins" / "disk-cleanup" 53 # Use the PluginManager's module naming convention so relative imports work. 54 spec = importlib.util.spec_from_file_location( 55 "hermes_plugins.disk_cleanup", 56 plugin_dir / "__init__.py", 57 submodule_search_locations=[str(plugin_dir)], 58 ) 59 # Ensure parent namespace package exists for the relative `. import disk_cleanup` 60 import types 61 if "hermes_plugins" not in sys.modules: 62 ns = types.ModuleType("hermes_plugins") 63 ns.__path__ = [] 64 sys.modules["hermes_plugins"] = ns 65 mod = importlib.util.module_from_spec(spec) 66 mod.__package__ = "hermes_plugins.disk_cleanup" 67 mod.__path__ = [str(plugin_dir)] 68 sys.modules["hermes_plugins.disk_cleanup"] = mod 69 spec.loader.exec_module(mod) 70 return mod 71 72 73 # --------------------------------------------------------------------------- 74 # Library tests 75 # --------------------------------------------------------------------------- 76 77 class TestIsSafePath: 78 def test_accepts_path_under_hermes_home(self, _isolate_env): 79 dg = _load_lib() 80 p = _isolate_env / "subdir" / "file.txt" 81 p.parent.mkdir() 82 p.write_text("x") 83 assert dg.is_safe_path(p) is True 84 85 def test_rejects_outside_hermes_home(self, _isolate_env): 86 dg = _load_lib() 87 assert dg.is_safe_path(Path("/etc/passwd")) is False 88 89 def test_accepts_tmp_hermes_prefix(self, _isolate_env, tmp_path): 90 dg = _load_lib() 91 assert dg.is_safe_path(Path("/tmp/hermes-abc/x.log")) is True 92 93 def test_rejects_plain_tmp(self, _isolate_env): 94 dg = _load_lib() 95 assert dg.is_safe_path(Path("/tmp/other.log")) is False 96 97 def test_rejects_windows_mount(self, _isolate_env): 98 dg = _load_lib() 99 assert dg.is_safe_path(Path("/mnt/c/Users/x/test.txt")) is False 100 101 102 class TestGuessCategory: 103 def test_test_prefix(self, _isolate_env): 104 dg = _load_lib() 105 p = _isolate_env / "test_foo.py" 106 p.write_text("x") 107 assert dg.guess_category(p) == "test" 108 109 def test_tmp_prefix(self, _isolate_env): 110 dg = _load_lib() 111 p = _isolate_env / "tmp_foo.log" 112 p.write_text("x") 113 assert dg.guess_category(p) == "test" 114 115 def test_dot_test_suffix(self, _isolate_env): 116 dg = _load_lib() 117 p = _isolate_env / "mything.test.js" 118 p.write_text("x") 119 assert dg.guess_category(p) == "test" 120 121 def test_skips_protected_top_level(self, _isolate_env): 122 dg = _load_lib() 123 logs_dir = _isolate_env / "logs" 124 logs_dir.mkdir() 125 p = logs_dir / "test_log.txt" 126 p.write_text("x") 127 # Even though it matches test_* pattern, logs/ is excluded. 128 assert dg.guess_category(p) is None 129 130 def test_cron_subtree_categorised(self, _isolate_env): 131 dg = _load_lib() 132 cron_dir = _isolate_env / "cron" 133 cron_dir.mkdir() 134 p = cron_dir / "job_output.md" 135 p.write_text("x") 136 assert dg.guess_category(p) == "cron-output" 137 138 def test_ordinary_file_returns_none(self, _isolate_env): 139 dg = _load_lib() 140 p = _isolate_env / "notes.md" 141 p.write_text("x") 142 assert dg.guess_category(p) is None 143 144 145 class TestTrackForgetQuick: 146 def test_track_then_quick_deletes_test(self, _isolate_env): 147 dg = _load_lib() 148 p = _isolate_env / "test_a.py" 149 p.write_text("x") 150 assert dg.track(str(p), "test", silent=True) is True 151 summary = dg.quick() 152 assert summary["deleted"] == 1 153 assert not p.exists() 154 155 def test_track_dedup(self, _isolate_env): 156 dg = _load_lib() 157 p = _isolate_env / "test_a.py" 158 p.write_text("x") 159 assert dg.track(str(p), "test", silent=True) is True 160 # Second call returns False (already tracked) 161 assert dg.track(str(p), "test", silent=True) is False 162 163 def test_track_rejects_outside_home(self, _isolate_env): 164 dg = _load_lib() 165 # /etc/hostname exists on most Linux boxes; fall back if not. 166 outside = "/etc/hostname" if Path("/etc/hostname").exists() else "/etc/passwd" 167 assert dg.track(outside, "test", silent=True) is False 168 169 def test_track_skips_missing(self, _isolate_env): 170 dg = _load_lib() 171 assert dg.track(str(_isolate_env / "nope.txt"), "test", silent=True) is False 172 173 def test_forget_removes_entry(self, _isolate_env): 174 dg = _load_lib() 175 p = _isolate_env / "keep.tmp" 176 p.write_text("x") 177 dg.track(str(p), "temp", silent=True) 178 assert dg.forget(str(p)) == 1 179 assert p.exists() # forget does NOT delete the file 180 181 def test_quick_preserves_unexpired_temp(self, _isolate_env): 182 dg = _load_lib() 183 p = _isolate_env / "fresh.tmp" 184 p.write_text("x") 185 dg.track(str(p), "temp", silent=True) 186 summary = dg.quick() 187 assert summary["deleted"] == 0 188 assert p.exists() 189 190 def test_quick_preserves_protected_top_level_dirs(self, _isolate_env): 191 dg = _load_lib() 192 for d in ("logs", "memories", "sessions", "cron", "cache"): 193 (_isolate_env / d).mkdir() 194 dg.quick() 195 for d in ("logs", "memories", "sessions", "cron", "cache"): 196 assert (_isolate_env / d).exists(), f"{d}/ should be preserved" 197 198 199 class TestStatus: 200 def test_empty_status(self, _isolate_env): 201 dg = _load_lib() 202 s = dg.status() 203 assert s["total_tracked"] == 0 204 assert s["top10"] == [] 205 206 def test_status_with_entries(self, _isolate_env): 207 dg = _load_lib() 208 p = _isolate_env / "big.tmp" 209 p.write_text("y" * 100) 210 dg.track(str(p), "temp", silent=True) 211 s = dg.status() 212 assert s["total_tracked"] == 1 213 assert len(s["top10"]) == 1 214 rendered = dg.format_status(s) 215 assert "temp" in rendered 216 assert "big.tmp" in rendered 217 218 219 class TestDryRun: 220 def test_classifies_by_category(self, _isolate_env): 221 dg = _load_lib() 222 test_f = _isolate_env / "test_x.py" 223 test_f.write_text("x") 224 big = _isolate_env / "big.bin" 225 big.write_bytes(b"z" * 10) 226 dg.track(str(test_f), "test", silent=True) 227 dg.track(str(big), "other", silent=True) 228 auto, prompt = dg.dry_run() 229 # test → auto, other → neither (doesn't hit any rule) 230 assert any(i["path"] == str(test_f) for i in auto) 231 232 233 # --------------------------------------------------------------------------- 234 # Plugin hooks tests 235 # --------------------------------------------------------------------------- 236 237 class TestPostToolCallHook: 238 def test_write_file_test_pattern_tracked(self, _isolate_env): 239 pi = _load_plugin_init() 240 p = _isolate_env / "test_created.py" 241 p.write_text("x") 242 pi._on_post_tool_call( 243 tool_name="write_file", 244 args={"path": str(p), "content": "x"}, 245 result="OK", 246 task_id="t1", session_id="s1", 247 ) 248 tracked_file = _isolate_env / "disk-cleanup" / "tracked.json" 249 data = json.loads(tracked_file.read_text()) 250 assert len(data) == 1 251 assert data[0]["category"] == "test" 252 253 def test_write_file_non_test_not_tracked(self, _isolate_env): 254 pi = _load_plugin_init() 255 p = _isolate_env / "notes.md" 256 p.write_text("x") 257 pi._on_post_tool_call( 258 tool_name="write_file", 259 args={"path": str(p), "content": "x"}, 260 result="OK", 261 task_id="t2", session_id="s2", 262 ) 263 tracked_file = _isolate_env / "disk-cleanup" / "tracked.json" 264 assert not tracked_file.exists() or tracked_file.read_text().strip() == "[]" 265 266 def test_terminal_command_picks_up_paths(self, _isolate_env): 267 pi = _load_plugin_init() 268 p = _isolate_env / "tmp_created.log" 269 p.write_text("x") 270 pi._on_post_tool_call( 271 tool_name="terminal", 272 args={"command": f"touch {p}"}, 273 result=f"created {p}\n", 274 task_id="t3", session_id="s3", 275 ) 276 tracked_file = _isolate_env / "disk-cleanup" / "tracked.json" 277 data = json.loads(tracked_file.read_text()) 278 assert any(Path(i["path"]) == p.resolve() for i in data) 279 280 def test_ignores_unrelated_tool(self, _isolate_env): 281 pi = _load_plugin_init() 282 pi._on_post_tool_call( 283 tool_name="read_file", 284 args={"path": str(_isolate_env / "test_x.py")}, 285 result="contents", 286 task_id="t4", session_id="s4", 287 ) 288 # read_file should never trigger tracking. 289 tracked_file = _isolate_env / "disk-cleanup" / "tracked.json" 290 assert not tracked_file.exists() or tracked_file.read_text().strip() == "[]" 291 292 293 class TestOnSessionEndHook: 294 def test_runs_quick_when_test_files_tracked(self, _isolate_env): 295 pi = _load_plugin_init() 296 p = _isolate_env / "test_cleanup.py" 297 p.write_text("x") 298 pi._on_post_tool_call( 299 tool_name="write_file", 300 args={"path": str(p), "content": "x"}, 301 result="OK", 302 task_id="", session_id="s1", 303 ) 304 assert p.exists() 305 pi._on_session_end(session_id="s1", completed=True, interrupted=False) 306 assert not p.exists(), "test file should be auto-deleted" 307 308 def test_noop_when_no_test_tracked(self, _isolate_env): 309 pi = _load_plugin_init() 310 # Nothing tracked → on_session_end should not raise. 311 pi._on_session_end(session_id="empty", completed=True, interrupted=False) 312 313 314 # --------------------------------------------------------------------------- 315 # Slash command 316 # --------------------------------------------------------------------------- 317 318 class TestSlashCommand: 319 def test_help(self, _isolate_env): 320 pi = _load_plugin_init() 321 out = pi._handle_slash("help") 322 assert "disk-cleanup" in out 323 assert "status" in out 324 325 def test_status_empty(self, _isolate_env): 326 pi = _load_plugin_init() 327 out = pi._handle_slash("status") 328 assert "nothing tracked" in out 329 330 def test_track_rejects_missing(self, _isolate_env): 331 pi = _load_plugin_init() 332 out = pi._handle_slash( 333 f"track {_isolate_env / 'nope.txt'} temp" 334 ) 335 assert "Not tracked" in out 336 337 def test_track_rejects_bad_category(self, _isolate_env): 338 pi = _load_plugin_init() 339 p = _isolate_env / "a.tmp" 340 p.write_text("x") 341 out = pi._handle_slash(f"track {p} banana") 342 assert "Unknown category" in out 343 344 def test_track_and_forget(self, _isolate_env): 345 pi = _load_plugin_init() 346 p = _isolate_env / "a.tmp" 347 p.write_text("x") 348 out = pi._handle_slash(f"track {p} temp") 349 assert "Tracked" in out 350 out = pi._handle_slash(f"forget {p}") 351 assert "Removed 1" in out 352 353 def test_unknown_subcommand(self, _isolate_env): 354 pi = _load_plugin_init() 355 out = pi._handle_slash("foobar") 356 assert "Unknown subcommand" in out 357 358 def test_quick_on_empty(self, _isolate_env): 359 pi = _load_plugin_init() 360 out = pi._handle_slash("quick") 361 assert "Cleaned 0 files" in out 362 363 364 # --------------------------------------------------------------------------- 365 # Bundled-plugin discovery 366 # --------------------------------------------------------------------------- 367 368 class TestBundledDiscovery: 369 def _write_enabled_config(self, hermes_home, names): 370 """Write plugins.enabled allow-list to config.yaml.""" 371 import yaml 372 cfg_path = hermes_home / "config.yaml" 373 cfg_path.write_text(yaml.safe_dump({"plugins": {"enabled": list(names)}})) 374 375 def test_disk_cleanup_discovered_but_not_loaded_by_default(self, _isolate_env): 376 """Bundled plugins are discovered but NOT loaded without opt-in.""" 377 from hermes_cli import plugins as pmod 378 mgr = pmod.PluginManager() 379 mgr.discover_and_load() 380 # Discovered — appears in the registry 381 assert "disk-cleanup" in mgr._plugins 382 loaded = mgr._plugins["disk-cleanup"] 383 assert loaded.manifest.source == "bundled" 384 # But NOT enabled — no hooks or commands registered 385 assert not loaded.enabled 386 assert loaded.error and "not enabled" in loaded.error 387 388 def test_disk_cleanup_loads_when_enabled(self, _isolate_env): 389 """Adding to plugins.enabled activates the bundled plugin.""" 390 self._write_enabled_config(_isolate_env, ["disk-cleanup"]) 391 from hermes_cli import plugins as pmod 392 mgr = pmod.PluginManager() 393 mgr.discover_and_load() 394 loaded = mgr._plugins["disk-cleanup"] 395 assert loaded.enabled 396 assert "post_tool_call" in loaded.hooks_registered 397 assert "on_session_end" in loaded.hooks_registered 398 assert "disk-cleanup" in loaded.commands_registered 399 400 def test_disabled_beats_enabled(self, _isolate_env): 401 """plugins.disabled wins even if the plugin is also in plugins.enabled.""" 402 import yaml 403 cfg_path = _isolate_env / "config.yaml" 404 cfg_path.write_text(yaml.safe_dump({ 405 "plugins": { 406 "enabled": ["disk-cleanup"], 407 "disabled": ["disk-cleanup"], 408 } 409 })) 410 from hermes_cli import plugins as pmod 411 mgr = pmod.PluginManager() 412 mgr.discover_and_load() 413 loaded = mgr._plugins["disk-cleanup"] 414 assert not loaded.enabled 415 assert loaded.error == "disabled via config" 416 417 def test_memory_and_context_engine_subdirs_skipped(self, _isolate_env): 418 """Bundled scan must NOT pick up plugins/memory or plugins/context_engine 419 as top-level plugins — they have their own discovery paths.""" 420 self._write_enabled_config( 421 _isolate_env, ["memory", "context_engine", "disk-cleanup"] 422 ) 423 from hermes_cli import plugins as pmod 424 mgr = pmod.PluginManager() 425 mgr.discover_and_load() 426 assert "memory" not in mgr._plugins 427 assert "context_engine" not in mgr._plugins