| 1 | from __future__ import annotations |
| 2 | |
| 3 | import os |
| 4 | import shutil |
| 5 | import uuid |
| 6 | from pathlib import Path |
| 7 | import sys |
| 8 | |
| 9 | import pytest |
| 10 | |
| 11 | PROJECT_ROOT = Path(__file__).resolve().parents[4] |
| 12 | if str(PROJECT_ROOT) not in sys.path: |
| 13 | sys.path.insert(0, str(PROJECT_ROOT)) |
| 14 | |
| 15 | from helpers import cache, files, plugins |
| 16 | from plugins._commands.api.commands import Commands |
| 17 | from plugins._commands.helpers import commands as commands_helper |
| 18 | |
| 19 | |
| 20 | # ── Fixtures ────────────────────────────────────────────────────────────────── |
| 21 | |
| 22 | @pytest.fixture(autouse=True) |
| 23 | def _clear_plugin_cache(): |
| 24 | """Ensure plugin list cache is fresh for every test.""" |
| 25 | cache.clear("*(plugins)*") |
| 26 | yield |
| 27 | cache.clear("*(plugins)*") |
| 28 | |
| 29 | |
| 30 | @pytest.fixture() |
| 31 | def fake_plugin(): |
| 32 | """Create a temporary plugin in usr/plugins/ with a commands/ directory.""" |
| 33 | suffix = uuid.uuid4().hex[:8] |
| 34 | plugin_name = f"_test_cmd_disc_{suffix}" |
| 35 | plugin_dir = files.get_abs_path(files.USER_DIR, files.PLUGINS_DIR, plugin_name) |
| 36 | commands_dir = os.path.join(plugin_dir, "commands") |
| 37 | os.makedirs(commands_dir, exist_ok=True) |
| 38 | |
| 39 | # Write plugin.yaml so the plugin is discoverable |
| 40 | files.write_file( |
| 41 | os.path.join(plugin_dir, "plugin.yaml"), |
| 42 | f"name: {plugin_name}\ntitle: Test\ndescription: Test\n", |
| 43 | ) |
| 44 | |
| 45 | yield {"name": plugin_name, "dir": plugin_dir, "commands_dir": commands_dir} |
| 46 | |
| 47 | # Cleanup |
| 48 | shutil.rmtree(plugin_dir, ignore_errors=True) |
| 49 | cache.remove(plugins.PLUGINS_LIST_CACHE_AREA, "") |
| 50 | |
| 51 | |
| 52 | def _write_plugin_command( |
| 53 | fake_plugin: dict, |
| 54 | *, |
| 55 | name: str, |
| 56 | description: str, |
| 57 | body: str = "default body", |
| 58 | ) -> str: |
| 59 | """Write a .command.yaml + .txt into the fake plugin's commands/ dir. |
| 60 | |
| 61 | Returns the config file path. |
| 62 | """ |
| 63 | slug = commands_helper.sanitize_command_name(name) |
| 64 | cdir = fake_plugin["commands_dir"] |
| 65 | config_path = os.path.join(cdir, f"{slug}.command.yaml") |
| 66 | content_path = os.path.join(cdir, f"{slug}.txt") |
| 67 | |
| 68 | files.write_file( |
| 69 | config_path, |
| 70 | f"name: {slug}\ndescription: {description}\ntype: text\ntemplate_path: {slug}.txt\n", |
| 71 | ) |
| 72 | files.write_file(content_path, body) |
| 73 | return config_path |
| 74 | |
| 75 | |
| 76 | # ── Tests ───────────────────────────────────────────────────────────────────── |
| 77 | |
| 78 | |
| 79 | def test_discover_plugin_commands_finds_plugin_commands(fake_plugin: dict): |
| 80 | """Plugin commands must be discovered without relying on real installs.""" |
| 81 | _write_plugin_command( |
| 82 | fake_plugin, |
| 83 | name=f"{fake_plugin['name']}-build", |
| 84 | description="build test", |
| 85 | ) |
| 86 | |
| 87 | discovered = commands_helper._discover_plugin_commands() |
| 88 | names = {c["name"] for c in discovered} |
| 89 | expected = commands_helper.sanitize_command_name(f"{fake_plugin['name']}-build") |
| 90 | assert expected in names, f"Expected {expected!r} in discovered names, got {names}" |
| 91 | |
| 92 | for cmd in discovered: |
| 93 | if cmd["name"] == expected: |
| 94 | assert cmd["source_plugin"] == fake_plugin["name"] |
| 95 | assert cmd["scope_key"] == "plugin" |
| 96 | assert cmd["scope_label"] == f"Plugin: {fake_plugin['name']}" |
| 97 | assert cmd["source_scope_key"] == "plugin" |
| 98 | assert cmd["source_scope_label"] == f"Plugin: {fake_plugin['name']}" |
| 99 | break |
| 100 | |
| 101 | |
| 102 | def test_discover_plugin_commands_skips_own_plugin(): |
| 103 | """The commands plugin itself must NOT appear in _discover_plugin_commands.""" |
| 104 | discovered = commands_helper._discover_plugin_commands() |
| 105 | for cmd in discovered: |
| 106 | assert cmd.get("source_plugin") != "_commands" |
| 107 | |
| 108 | |
| 109 | def test_discover_plugin_commands_skips_disabled_plugins(fake_plugin: dict): |
| 110 | """Disabled plugins must not contribute slash commands to the picker.""" |
| 111 | _write_plugin_command( |
| 112 | fake_plugin, |
| 113 | name=f"{fake_plugin['name']}-disabled", |
| 114 | description="disabled command", |
| 115 | ) |
| 116 | files.write_file(os.path.join(fake_plugin["dir"], plugins.DISABLED_FILE_NAME), "") |
| 117 | cache.clear("*(plugins)*") |
| 118 | |
| 119 | discovered = commands_helper._discover_plugin_commands() |
| 120 | names = {command["name"] for command in discovered} |
| 121 | expected = commands_helper.sanitize_command_name(f"{fake_plugin['name']}-disabled") |
| 122 | assert expected not in names |
| 123 | |
| 124 | |
| 125 | def test_discover_builtin_commands_marks_own_commands_read_only(): |
| 126 | """Bundled _commands command files are discoverable as built-ins, not plugin commands.""" |
| 127 | discovered = commands_helper._discover_builtin_commands() |
| 128 | command = next((cmd for cmd in discovered if cmd["name"] == "new"), None) |
| 129 | |
| 130 | assert command is not None |
| 131 | assert command["source_plugin"] == "_commands" |
| 132 | assert command["scope_key"] == "builtin" |
| 133 | assert command["scope_label"] == "Built-in" |
| 134 | |
| 135 | loaded = commands_helper.get_command(command["path"]) |
| 136 | assert loaded["name"] == "new" |
| 137 | assert loaded["scope_key"] == "builtin" |
| 138 | |
| 139 | with pytest.raises(ValueError, match="Built-in commands are read-only"): |
| 140 | commands_helper.save_command( |
| 141 | existing_path=command["path"], |
| 142 | name="new", |
| 143 | description="updated description", |
| 144 | body="updated body", |
| 145 | ) |
| 146 | |
| 147 | with pytest.raises(ValueError, match="Built-in commands are read-only"): |
| 148 | commands_helper.delete_command(command["path"]) |
| 149 | |
| 150 | response = object.__new__(Commands)._list_scope({"project_name": ""}) |
| 151 | assert "new" in {item["name"] for item in response["builtin_commands"]} |
| 152 | |
| 153 | |
| 154 | def test_builtin_commands_use_canonical_names_only(): |
| 155 | discovered = commands_helper._discover_builtin_commands() |
| 156 | names = {command["name"] for command in discovered} |
| 157 | |
| 158 | assert { |
| 159 | "attach", |
| 160 | "computer-use", |
| 161 | "models", |
| 162 | "permissions", |
| 163 | "plugins", |
| 164 | "project", |
| 165 | "stop", |
| 166 | } <= names |
| 167 | assert { |
| 168 | "computer", |
| 169 | "cu", |
| 170 | "disconnect", |
| 171 | "exit", |
| 172 | "help", |
| 173 | "image", |
| 174 | "img", |
| 175 | "keys", |
| 176 | "model", |
| 177 | "plugin", |
| 178 | "projects", |
| 179 | }.isdisjoint(names) |
| 180 | |
| 181 | |
| 182 | def test_webui_effective_list_hides_webui_hidden_commands(): |
| 183 | effective, _ = commands_helper.list_effective_commands("") |
| 184 | chats = next(command for command in effective if command["name"] == "chats") |
| 185 | response = object.__new__(Commands)._list_effective({"context_id": ""}) |
| 186 | names = {command["name"] for command in response["commands"]} |
| 187 | |
| 188 | assert chats["frontmatter_extra"]["webui_hidden"] is True |
| 189 | assert "chats" not in names |
| 190 | |
| 191 | |
| 192 | def test_list_effective_includes_plugin_commands(fake_plugin: dict): |
| 193 | """list_effective_commands must include commands from other plugins.""" |
| 194 | _write_plugin_command( |
| 195 | fake_plugin, |
| 196 | name=f"{fake_plugin['name']}-effective", |
| 197 | description="effective test", |
| 198 | ) |
| 199 | |
| 200 | effective, _ = commands_helper.list_effective_commands("") |
| 201 | expected = commands_helper.sanitize_command_name(f"{fake_plugin['name']}-effective") |
| 202 | command = next((item for item in effective if item["name"] == expected), None) |
| 203 | assert command is not None |
| 204 | assert command["scope_key"] == "plugin" |
| 205 | assert command["scope_label"] == f"Plugin: {fake_plugin['name']}" |
| 206 | |
| 207 | |
| 208 | def test_plugin_commands_appear_in_effective_list(fake_plugin: dict): |
| 209 | """Commands from a freshly-created plugin appear in effective list.""" |
| 210 | _write_plugin_command( |
| 211 | fake_plugin, |
| 212 | name=f"{fake_plugin['name']}-hello", |
| 213 | description="A test command from a plugin", |
| 214 | body="Hello from plugin", |
| 215 | ) |
| 216 | |
| 217 | effective, _ = commands_helper.list_effective_commands("") |
| 218 | expected = commands_helper.sanitize_command_name(f"{fake_plugin['name']}-hello") |
| 219 | command = next((item for item in effective if item["name"] == expected), None) |
| 220 | assert command is not None |
| 221 | assert command["source_plugin"] == fake_plugin["name"] |
| 222 | |
| 223 | |
| 224 | def test_precedence_global_overrides_plugin(fake_plugin: dict): |
| 225 | """A global command with the same name takes precedence over a plugin command.""" |
| 226 | shared_name = f"{fake_plugin['name']}-shared" |
| 227 | slug = commands_helper.sanitize_command_name(shared_name) |
| 228 | |
| 229 | # 1. Plugin command (lowest precedence) |
| 230 | _write_plugin_command( |
| 231 | fake_plugin, |
| 232 | name=shared_name, |
| 233 | description="plugin version", |
| 234 | body="plugin body", |
| 235 | ) |
| 236 | |
| 237 | # 2. Global command (higher precedence) |
| 238 | try: |
| 239 | commands_helper.save_command( |
| 240 | name=shared_name, |
| 241 | description="global version", |
| 242 | body="global body", |
| 243 | ) |
| 244 | |
| 245 | effective, _ = commands_helper.list_effective_commands("") |
| 246 | by_name = {c["name"]: c for c in effective} |
| 247 | assert slug in by_name |
| 248 | assert by_name[slug]["description"] == "global version" |
| 249 | finally: |
| 250 | scope_dir = commands_helper.get_scope_directory("") |
| 251 | files.delete_file(os.path.join(scope_dir, f"{slug}.command.yaml")) |
| 252 | files.delete_file(os.path.join(scope_dir, f"{slug}.txt")) |
| 253 | |
| 254 | |
| 255 | def test_source_plugin_field_on_discovered_command(fake_plugin: dict): |
| 256 | """Discovered commands must carry the source_plugin field.""" |
| 257 | _write_plugin_command( |
| 258 | fake_plugin, |
| 259 | name=f"{fake_plugin['name']}-src-test", |
| 260 | description="source plugin test", |
| 261 | ) |
| 262 | |
| 263 | discovered = commands_helper._discover_plugin_commands() |
| 264 | slug = commands_helper.sanitize_command_name(f"{fake_plugin['name']}-src-test") |
| 265 | match = next((c for c in discovered if c["name"] == slug), None) |
| 266 | assert match is not None |
| 267 | assert match["source_plugin"] == fake_plugin["name"] |
| 268 | assert match["scope_key"] == "plugin" |
| 269 | assert match["scope_label"] == f"Plugin: {fake_plugin['name']}" |
| 270 | assert match["source_scope_key"] == "plugin" |
| 271 | assert match["source_scope_label"] == f"Plugin: {fake_plugin['name']}" |
| 272 | |
| 273 | |
| 274 | def test_is_plugin_commands_dir_recognises_plugin_path(fake_plugin: dict): |
| 275 | """_is_plugin_commands_dir must return True for files inside plugin commands/ dirs.""" |
| 276 | _write_plugin_command( |
| 277 | fake_plugin, |
| 278 | name=f"{fake_plugin['name']}-path-check", |
| 279 | description="path test", |
| 280 | ) |
| 281 | slug = commands_helper.sanitize_command_name(f"{fake_plugin['name']}-path-check") |
| 282 | config_path = os.path.join(fake_plugin["commands_dir"], f"{slug}.command.yaml") |
| 283 | normalized = commands_helper._normalize_client_path(config_path) |
| 284 | |
| 285 | assert commands_helper._is_plugin_commands_dir(normalized) is True |
| 286 | |
| 287 | |
| 288 | def test_is_plugin_commands_dir_rejects_non_plugin_path(tmp_path: Path): |
| 289 | """_is_plugin_commands_dir must return False for arbitrary paths.""" |
| 290 | non_plugin_path = tmp_path / "not-a-plugin" / "commands" / "foo.txt" |
| 291 | assert commands_helper._is_plugin_commands_dir(str(non_plugin_path)) is False |
| 292 | |
| 293 | |
| 294 | def test_get_command_can_load_plugin_command(fake_plugin: dict): |
| 295 | """get_command must work for commands inside plugin directories.""" |
| 296 | _write_plugin_command( |
| 297 | fake_plugin, |
| 298 | name=f"{fake_plugin['name']}-loadable", |
| 299 | description="loadable test", |
| 300 | body="load me", |
| 301 | ) |
| 302 | slug = commands_helper.sanitize_command_name(f"{fake_plugin['name']}-loadable") |
| 303 | config_path = os.path.join(fake_plugin["commands_dir"], f"{slug}.command.yaml") |
| 304 | normalized = commands_helper._normalize_client_path(config_path) |
| 305 | |
| 306 | command = commands_helper.get_command(normalized, project_name="demo-project") |
| 307 | assert command["name"] == slug |
| 308 | assert command["description"] == "loadable test" |
| 309 | assert command["body"] == "load me" |
| 310 | assert command["source_plugin"] == fake_plugin["name"] |
| 311 | assert command["scope_key"] == "plugin" |
| 312 | assert command["scope_label"] == f"Plugin: {fake_plugin['name']}" |
| 313 | assert command["source_scope_key"] == "plugin" |
| 314 | assert command["source_scope_label"] == f"Plugin: {fake_plugin['name']}" |
| 315 | |
| 316 | |
| 317 | def test_save_command_rejects_plugin_existing_path(fake_plugin: dict): |
| 318 | """Editing a plugin command must fail because plugin commands are read-only.""" |
| 319 | config_path = _write_plugin_command( |
| 320 | fake_plugin, |
| 321 | name=f"{fake_plugin['name']}-readonly-edit", |
| 322 | description="read-only test", |
| 323 | body="plugin body", |
| 324 | ) |
| 325 | normalized = commands_helper._normalize_client_path(config_path) |
| 326 | |
| 327 | with pytest.raises(ValueError, match="Plugin commands are read-only"): |
| 328 | commands_helper.save_command( |
| 329 | existing_path=normalized, |
| 330 | name=f"{fake_plugin['name']}-readonly-edit", |
| 331 | description="updated description", |
| 332 | body="updated body", |
| 333 | ) |
| 334 | |
| 335 | |
| 336 | def test_delete_command_rejects_plugin_command(fake_plugin: dict): |
| 337 | """Deleting a plugin command must fail because plugin commands are read-only.""" |
| 338 | config_path = _write_plugin_command( |
| 339 | fake_plugin, |
| 340 | name=f"{fake_plugin['name']}-readonly-delete", |
| 341 | description="read-only delete", |
| 342 | ) |
| 343 | normalized = commands_helper._normalize_client_path(config_path) |
| 344 | |
| 345 | with pytest.raises(ValueError, match="Plugin commands are read-only"): |
| 346 | commands_helper.delete_command(normalized) |
| 347 | |
| 348 | assert os.path.exists(config_path) |