"""Tests for skills and hooks auto-install.""" import json import os import subprocess import stat import sys from pathlib import Path from unittest.mock import patch import pytest if sys.version_info >= (3, 11): import tomllib else: # pragma: no cover - Python 3.10 backport import tomli as tomllib from code_review_graph.skills import ( _CLAUDE_MD_SECTION_MARKER, PLATFORMS, _cursor_hook_scripts, _detect_serve_command, _in_poetry_project, _in_uv_project, _opencode_plugin_content, generate_codex_hooks_config, generate_cursor_hooks_config, generate_hooks_config, generate_skills, inject_claude_md, inject_platform_instructions, install_codex_hooks, install_gemini_cli_hooks, install_gemini_cli_skills, install_cursor_hooks, install_git_hook, install_hooks, install_opencode_plugin, install_platform_configs, ) _needs_tomllib = pytest.mark.skipif( tomllib is None, reason="tomllib requires Python 3.11+", ) class TestGenerateSkills: def test_creates_skills_directory(self, tmp_path): result = generate_skills(tmp_path) assert result.is_dir() assert result == tmp_path / ".claude" / "skills" def test_creates_four_skill_subdirs(self, tmp_path): skills_dir = generate_skills(tmp_path) subdirs = sorted(f.name for f in skills_dir.iterdir() if f.is_dir()) assert subdirs == [ "debug-issue", "explore-codebase", "refactor-safely", "review-changes", ] for d in skills_dir.iterdir(): assert (d / "skill.md").is_file() def test_skill_files_have_frontmatter(self, tmp_path): skills_dir = generate_skills(tmp_path) for subdir in skills_dir.iterdir(): path = subdir / "skill.md" content = path.read_text() assert content.startswith("---\n") assert "name:" in content assert "description:" in content # Frontmatter closes lines = content.split("\n") assert lines[0] == "---" closing_idx = content.index("---", 4) assert closing_idx > 0 def test_custom_skills_dir(self, tmp_path): custom = tmp_path / "my-skills" result = generate_skills(tmp_path, skills_dir=custom) assert result == custom assert result.is_dir() assert len(list(result.iterdir())) == 4 def test_skill_content_includes_get_minimal_context(self, tmp_path): """Every skill template must reference get_minimal_context.""" skills_dir = generate_skills(tmp_path) for subdir in skills_dir.iterdir(): content = (subdir / "skill.md").read_text() assert "get_minimal_context" in content, ( f"{subdir.name} missing get_minimal_context reference" ) def test_skill_content_includes_detail_level(self, tmp_path): """Every skill template must reference detail_level.""" skills_dir = generate_skills(tmp_path) for subdir in skills_dir.iterdir(): content = (subdir / "skill.md").read_text() assert "detail_level" in content, ( f"{subdir.name} missing detail_level reference" ) def test_idempotent(self, tmp_path): """Running twice should not fail and files should still be valid.""" generate_skills(tmp_path) generate_skills(tmp_path) skills_dir = tmp_path / ".claude" / "skills" assert len(list(skills_dir.iterdir())) == 4 class TestGenerateHooksConfig: def test_returns_dict_with_hooks(self): config = generate_hooks_config(Path("/repo")) assert "hooks" in config def test_has_post_tool_use(self): config = generate_hooks_config(Path("/repo")) assert "PostToolUse" in config["hooks"] entry = config["hooks"]["PostToolUse"][0] assert entry["matcher"] == "Edit|Write|Bash" inner = entry["hooks"][0] assert inner["type"] == "command" assert "update" in inner["command"] assert inner["command"].startswith("cat >/dev/null || true; ") assert 0 < inner["timeout"] <= 600 def test_has_session_start(self): config = generate_hooks_config(Path("/repo")) assert "SessionStart" in config["hooks"] entry = config["hooks"]["SessionStart"][0] assert "matcher" in entry inner = entry["hooks"][0] assert inner["type"] == "command" assert "status" in inner["command"] assert inner["command"].startswith("cat >/dev/null || true; ") assert 0 < inner["timeout"] <= 600 def test_does_not_emit_invalid_pre_commit_hook(self): config = generate_hooks_config(Path("/repo")) assert "PreCommit" not in config["hooks"] def test_has_only_valid_hook_types(self): config = generate_hooks_config(Path("/repo")) hook_types = set(config["hooks"].keys()) assert hook_types == {"PostToolUse", "SessionStart"} def test_hook_entries_use_nested_hooks_array(self): config = generate_hooks_config(Path("/repo")) for hook_type, entries in config["hooks"].items(): for entry in entries: assert "hooks" in entry, f"{hook_type} entry missing 'hooks' array" assert "command" not in entry, f"{hook_type} has bare 'command' outside hooks[]" def test_repo_root_embedded_in_commands(self): config = generate_hooks_config(Path("/my/project")) post_cmd = config["hooks"]["PostToolUse"][0]["hooks"][0]["command"] session_cmd = config["hooks"]["SessionStart"][0]["hooks"][0]["command"] assert "/my/project" in post_cmd assert "/my/project" in session_cmd def test_quotes_repo_paths_with_spaces(self): config = generate_hooks_config(Path("/repo with spaces")) post_cmd = config["hooks"]["PostToolUse"][0]["hooks"][0]["command"] assert '"' in post_cmd # path is JSON-encoded so spaces are quoted def test_entries_use_claude_code_hook_schema(self): """Regression guard for the Claude Code hook schema. Claude Code rejects entries that put ``command`` directly on the event entry. Each entry must wrap its command(s) in a ``hooks: [{"type": "command", "command": ..., "timeout": ...}]`` array — missing that wrapper causes the entire settings.json to fail to parse ("Expected array, but received undefined"). """ config = generate_hooks_config(Path("/repo")) for event_name, entries in config["hooks"].items(): for entry in entries: assert "command" not in entry, ( f"{event_name} entry has a flat `command` field; " "it must be wrapped in an inner `hooks` array" ) assert "hooks" in entry, ( f"{event_name} entry is missing the inner `hooks` array" ) assert isinstance(entry["hooks"], list) for hook in entry["hooks"]: assert hook.get("type") == "command", ( f"{event_name} inner hook missing type=\"command\"" ) assert "command" in hook assert "timeout" in hook class TestShippedHooksFiles: """The vestigial hooks/ directory ships in the sdist (see pyproject sdist includes). Its hook commands must drain stdin exactly like the skills.py-generated hooks, or large hook payloads reproduce the BrokenPipeError from bug #493. """ HOOKS_DIR = Path(__file__).resolve().parent.parent / "hooks" STDIN_DRAIN = "cat >/dev/null || true; " def test_hooks_json_commands_drain_stdin(self): data = json.loads( (self.HOOKS_DIR / "hooks.json").read_text(encoding="utf-8") ) commands = [ hook["command"] for entries in data.values() for entry in entries for hook in entry.get("hooks", []) if hook.get("type") == "command" ] assert commands, "hooks/hooks.json should define at least one command hook" for command in commands: assert command.startswith(self.STDIN_DRAIN), ( f"hooks.json command lacks the stdin drain prefix: {command!r}" ) def test_session_start_script_drains_stdin(self): script = (self.HOOKS_DIR / "session-start.sh").read_text(encoding="utf-8") assert "cat >/dev/null" in script, ( "session-start.sh must drain stdin to avoid BrokenPipeError " "on large hook payloads (bug #493)" ) class TestInstallGitHook: def _make_git_repo(self, tmp_path: Path) -> Path: (tmp_path / ".git" / "hooks").mkdir(parents=True) return tmp_path def _git(self, *args: str, cwd: Path) -> str: result = subprocess.run( ["git", *args], cwd=str(cwd), capture_output=True, text=True, encoding="utf-8", stdin=subprocess.DEVNULL, timeout=30, check=True, ) return result.stdout.strip() def _init_real_repo(self, path: Path) -> Path: path.mkdir(parents=True, exist_ok=True) self._git("init", cwd=path) return path def test_creates_executable_pre_commit_hook(self, tmp_path): hook_path = install_git_hook(self._make_git_repo(tmp_path)) assert hook_path is not None and hook_path.name == "pre-commit" assert os.access(hook_path, os.X_OK) content = hook_path.read_text() assert content.startswith("#!/") assert "code-review-graph detect-changes" in content def test_appends_to_existing_hook(self, tmp_path): repo = self._make_git_repo(tmp_path) hook_path = repo / ".git" / "hooks" / "pre-commit" hook_path.write_text("#!/bin/sh\nexisting-command\n", encoding="utf-8") hook_path.chmod(0o755) install_git_hook(repo) content = hook_path.read_text() assert "existing-command" in content assert "code-review-graph detect-changes" in content def test_idempotent(self, tmp_path): repo = self._make_git_repo(tmp_path) install_git_hook(repo) install_git_hook(repo) content = (repo / ".git" / "hooks" / "pre-commit").read_text() assert content.count("code-review-graph detect-changes") == 1 def test_no_git_dir_returns_none(self, tmp_path): assert install_git_hook(tmp_path) is None def test_real_repo_installs_into_git_hooks(self, tmp_path): """Standard repo: unchanged behavior — hook lands in .git/hooks.""" repo = self._init_real_repo(tmp_path / "std") hook_path = install_git_hook(repo) assert hook_path is not None expected = repo / ".git" / "hooks" / "pre-commit" assert hook_path.resolve() == expected.resolve() assert os.access(hook_path, os.X_OK) assert "code-review-graph detect-changes" in hook_path.read_text() def test_respects_core_hooks_path(self, tmp_path): """core.hooksPath (husky-style): the hook must land where git runs it.""" repo = self._init_real_repo(tmp_path / "husky") self._git("config", "core.hooksPath", ".husky", cwd=repo) hook_path = install_git_hook(repo) assert hook_path is not None expected = repo / ".husky" / "pre-commit" assert hook_path.resolve() == expected.resolve() assert os.access(hook_path, os.X_OK) assert "code-review-graph detect-changes" in hook_path.read_text() # The default location must NOT be used — git would never run it. assert not (repo / ".git" / "hooks" / "pre-commit").exists() def test_linked_worktree_installs_where_git_runs_hooks(self, tmp_path): """Linked worktree: .git is a file; the hook must still be installed into the hooks path git actually consults (issue #313).""" main = self._init_real_repo(tmp_path / "main") self._git( "-c", "user.email=test@example.com", "-c", "user.name=Test", "commit", "--allow-empty", "-m", "init", cwd=main, ) worktree = tmp_path / "wt" self._git("worktree", "add", str(worktree), "-b", "wt-branch", cwd=main) assert (worktree / ".git").is_file() # precondition: not a directory hook_path = install_git_hook(worktree) assert hook_path is not None git_hooks_dir = worktree / self._git( "rev-parse", "--git-path", "hooks", cwd=worktree ) assert hook_path.resolve() == (git_hooks_dir / "pre-commit").resolve() assert "code-review-graph detect-changes" in hook_path.read_text() class TestInstallHooks: def test_creates_settings_file(self, tmp_path): install_hooks(tmp_path) settings_path = tmp_path / ".claude" / "settings.json" assert settings_path.exists() data = json.loads(settings_path.read_text()) assert "hooks" in data def test_merges_with_existing(self, tmp_path): settings_dir = tmp_path / ".claude" settings_dir.mkdir(parents=True) existing = {"customSetting": True, "hooks": {"OtherHook": []}} (settings_dir / "settings.json").write_text(json.dumps(existing)) install_hooks(tmp_path) data = json.loads((settings_dir / "settings.json").read_text()) assert data["customSetting"] is True assert "OtherHook" in data["hooks"] assert "PostToolUse" in data["hooks"] assert "SessionStart" in data["hooks"] assert "PreCommit" not in data["hooks"] assert "OtherHook" in data["hooks"] # pre-existing hooks must not be clobbered def test_creates_settings_backup(self, tmp_path): settings_dir = tmp_path / ".claude" settings_dir.mkdir(parents=True) existing = {"hooks": {"OtherHook": []}} (settings_dir / "settings.json").write_text(json.dumps(existing)) install_hooks(tmp_path) backup_path = settings_dir / "settings.json.bak" assert backup_path.exists() backup = json.loads(backup_path.read_text()) assert backup == existing def test_creates_claude_directory(self, tmp_path): install_hooks(tmp_path) assert (tmp_path / ".claude").is_dir() class TestGenerateCodexHooksConfig: def test_returns_dict_with_hooks(self, tmp_path): config = generate_codex_hooks_config(tmp_path) assert "hooks" in config def test_has_post_tool_use(self, tmp_path): config = generate_codex_hooks_config(tmp_path) assert "PostToolUse" in config["hooks"] entry = config["hooks"]["PostToolUse"][0] assert entry["matcher"] == "Write|Edit|Bash" inner = entry["hooks"][0] assert inner["type"] == "command" assert "update" in inner["command"] assert inner["command"].startswith("cat >/dev/null || true; ") assert inner["statusMessage"] == "Updating code-review-graph" def test_has_session_start(self, tmp_path): config = generate_codex_hooks_config(tmp_path) assert "SessionStart" in config["hooks"] entry = config["hooks"]["SessionStart"][0] assert entry["matcher"] == "startup|resume" inner = entry["hooks"][0] assert inner["type"] == "command" assert "status" in inner["command"] assert inner["command"].startswith("cat >/dev/null || true; ") assert inner["statusMessage"] == "Checking code-review-graph status" def test_post_tool_use_command_handles_large_stdin_payload(self, tmp_path): config = generate_codex_hooks_config(tmp_path) cmd = config["hooks"]["PostToolUse"][0]["hooks"][0]["command"] payload = ("x" * 1024 + "\n") * 20000 proc = subprocess.Popen( ["bash", "-lc", cmd], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=tmp_path, ) broken_pipe = None try: assert proc.stdin is not None proc.stdin.write(payload) proc.stdin.close() except BrokenPipeError as exc: # pragma: no cover - regression guard broken_pipe = exc proc.stdin = None stdout, stderr = proc.communicate() assert broken_pipe is None, f"hook command raised BrokenPipeError: {stderr}" assert proc.returncode == 0, stderr def test_commands_do_not_pin_a_specific_repo_path(self, tmp_path): config = generate_codex_hooks_config(tmp_path / "repo with spaces") post_cmd = config["hooks"]["PostToolUse"][0]["hooks"][0]["command"] session_cmd = config["hooks"]["SessionStart"][0]["hooks"][0]["command"] assert "--repo" not in post_cmd assert "--repo" not in session_cmd assert "code-review-graph update --skip-flows" in post_cmd assert "code-review-graph status" in session_cmd class TestInstallCodexHooks: def test_creates_hooks_file(self, tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) hooks_path = install_codex_hooks(tmp_path / "repo") assert hooks_path == tmp_path / ".codex" / "hooks.json" assert hooks_path.exists() data = json.loads(hooks_path.read_text()) assert "hooks" in data assert "PostToolUse" in data["hooks"] assert "SessionStart" in data["hooks"] def test_merges_with_existing(self, tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) codex_dir = tmp_path / ".codex" codex_dir.mkdir(parents=True) existing = { "customSetting": True, "hooks": { "Stop": [{"hooks": [{"type": "command", "command": "echo stop"}]}], }, } (codex_dir / "hooks.json").write_text(json.dumps(existing), encoding="utf-8") install_codex_hooks(tmp_path / "repo") data = json.loads((codex_dir / "hooks.json").read_text()) assert data["customSetting"] is True assert "Stop" in data["hooks"] assert "PostToolUse" in data["hooks"] assert "SessionStart" in data["hooks"] def test_creates_hooks_backup(self, tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) codex_dir = tmp_path / ".codex" codex_dir.mkdir(parents=True) existing = {"hooks": {"Stop": []}} hooks_path = codex_dir / "hooks.json" hooks_path.write_text(json.dumps(existing), encoding="utf-8") install_codex_hooks(tmp_path / "repo") backup_path = codex_dir / "hooks.json.bak" assert backup_path.exists() backup = json.loads(backup_path.read_text()) assert backup == existing def test_idempotent_by_command(self, tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) repo_root = tmp_path / "repo" install_codex_hooks(repo_root) install_codex_hooks(repo_root) data = json.loads((tmp_path / ".codex" / "hooks.json").read_text()) assert len(data["hooks"]["PostToolUse"]) == 1 assert len(data["hooks"]["SessionStart"]) == 1 def test_install_qoder_hooks(self, tmp_path): install_hooks(tmp_path, platform="qoder") settings_path = tmp_path / ".qoder" / "settings.json" assert settings_path.exists() data = json.loads(settings_path.read_text()) assert "hooks" in data assert "PostToolUse" in data["hooks"] assert "SessionStart" in data["hooks"] def test_install_qoder_hooks_merges_existing(self, tmp_path): settings_dir = tmp_path / ".qoder" settings_dir.mkdir(parents=True) existing = {"customSetting": True} (settings_dir / "settings.json").write_text(json.dumps(existing)) install_hooks(tmp_path, platform="qoder") data = json.loads((settings_dir / "settings.json").read_text()) assert data["customSetting"] is True assert "hooks" in data class TestInjectClaudeMd: def test_creates_section_in_new_file(self, tmp_path): inject_claude_md(tmp_path) content = (tmp_path / "CLAUDE.md").read_text() assert _CLAUDE_MD_SECTION_MARKER in content assert "MCP Tools" in content def test_appends_to_existing_file(self, tmp_path): claude_md = tmp_path / "CLAUDE.md" claude_md.write_text("# My Project\n\nExisting content.\n") inject_claude_md(tmp_path) content = claude_md.read_text() assert "# My Project" in content assert "Existing content." in content assert _CLAUDE_MD_SECTION_MARKER in content def test_idempotent(self, tmp_path): """Running twice should not duplicate the section.""" inject_claude_md(tmp_path) first_content = (tmp_path / "CLAUDE.md").read_text() inject_claude_md(tmp_path) second_content = (tmp_path / "CLAUDE.md").read_text() assert first_content == second_content assert second_content.count(_CLAUDE_MD_SECTION_MARKER) == 1 def test_idempotent_with_existing_content(self, tmp_path): claude_md = tmp_path / "CLAUDE.md" claude_md.write_text("# Existing\n") inject_claude_md(tmp_path) first_content = claude_md.read_text() inject_claude_md(tmp_path) second_content = claude_md.read_text() assert first_content == second_content assert second_content.count(_CLAUDE_MD_SECTION_MARKER) == 1 class TestInjectPlatformInstructionsFiltering: def test_all_writes_every_file(self, tmp_path): updated = inject_platform_instructions(tmp_path, target="all") assert set(updated) == { "AGENTS.md", "GEMINI.md", ".cursorrules", ".windsurfrules", "QODER.md", ".kiro/steering/code-review-graph.md", ".github/code-review-graph.instruction.md", } def test_default_is_all(self, tmp_path): updated = inject_platform_instructions(tmp_path) assert set(updated) == { "AGENTS.md", "GEMINI.md", ".cursorrules", ".windsurfrules", "QODER.md", ".kiro/steering/code-review-graph.md", ".github/code-review-graph.instruction.md", } def test_claude_writes_nothing(self, tmp_path): updated = inject_platform_instructions(tmp_path, target="claude") assert updated == [] assert not (tmp_path / "AGENTS.md").exists() assert not (tmp_path / "GEMINI.md").exists() assert not (tmp_path / ".cursorrules").exists() assert not (tmp_path / ".windsurfrules").exists() assert not (tmp_path / "QODER.md").exists() assert not (tmp_path / ".github" / "code-review-graph.instruction.md").exists() def test_cursor_writes_only_cursor_files(self, tmp_path): updated = inject_platform_instructions(tmp_path, target="cursor") assert set(updated) == {"AGENTS.md", ".cursorrules"} assert not (tmp_path / "GEMINI.md").exists() assert not (tmp_path / ".windsurfrules").exists() assert not (tmp_path / "QODER.md").exists() def test_windsurf_writes_only_windsurfrules(self, tmp_path): updated = inject_platform_instructions(tmp_path, target="windsurf") assert updated == [".windsurfrules"] def test_antigravity_writes_agents_and_gemini(self, tmp_path): updated = inject_platform_instructions(tmp_path, target="antigravity") assert set(updated) == {"AGENTS.md", "GEMINI.md"} def test_gemini_cli_writes_only_gemini_md(self, tmp_path): updated = inject_platform_instructions(tmp_path, target="gemini-cli") assert updated == ["GEMINI.md"] assert not (tmp_path / "AGENTS.md").exists() assert not (tmp_path / ".cursorrules").exists() assert not (tmp_path / ".windsurfrules").exists() assert not (tmp_path / "QODER.md").exists() def test_opencode_writes_only_agents(self, tmp_path): updated = inject_platform_instructions(tmp_path, target="opencode") assert updated == ["AGENTS.md"] def test_qoder_writes_only_qoder_md(self, tmp_path): updated = inject_platform_instructions(tmp_path, target="qoder") assert updated == ["QODER.md"] assert not (tmp_path / "AGENTS.md").exists() assert not (tmp_path / "GEMINI.md").exists() assert not (tmp_path / ".cursorrules").exists() assert not (tmp_path / ".windsurfrules").exists() class TestInstallPlatformConfigs: @_needs_tomllib def test_install_codex_config(self, tmp_path): codex_config = tmp_path / ".codex" / "config.toml" with patch.dict( PLATFORMS, { "codex": { **PLATFORMS["codex"], "config_path": lambda root: codex_config, "detect": lambda: True, }, }, ): configured = install_platform_configs(tmp_path, target="codex") assert "Codex" in configured data = tomllib.loads(codex_config.read_text()) entry = data["mcp_servers"]["code-review-graph"] assert entry["type"] == "stdio" assert "serve" in entry["args"] @_needs_tomllib def test_install_codex_preserves_existing_toml(self, tmp_path): codex_config = tmp_path / ".codex" / "config.toml" codex_config.parent.mkdir(parents=True) codex_config.write_text( 'model = "gpt-5.4"\n\n[mcp_servers.other]\ncommand = "other"\n', encoding="utf-8", ) with patch.dict( PLATFORMS, { "codex": { **PLATFORMS["codex"], "config_path": lambda root: codex_config, "detect": lambda: True, }, }, ): install_platform_configs(tmp_path, target="codex") data = tomllib.loads(codex_config.read_text()) assert data["model"] == "gpt-5.4" assert data["mcp_servers"]["other"]["command"] == "other" expected_cmd, _ = _detect_serve_command() assert data["mcp_servers"]["code-review-graph"]["command"] == expected_cmd def test_install_codex_no_duplicate(self, tmp_path): codex_config = tmp_path / ".codex" / "config.toml" codex_config.parent.mkdir(parents=True) codex_config.write_text( "\n".join( [ "[mcp_servers.code-review-graph]", 'command = "uvx"', 'args = ["code-review-graph", "serve"]', 'type = "stdio"', "", ] ), encoding="utf-8", ) with patch.dict( PLATFORMS, { "codex": { **PLATFORMS["codex"], "config_path": lambda root: codex_config, "detect": lambda: True, }, }, ): install_platform_configs(tmp_path, target="codex") assert codex_config.read_text().count("[mcp_servers.code-review-graph]") == 1 def test_install_cursor_config(self, tmp_path): with patch.dict( PLATFORMS, { "cursor": {**PLATFORMS["cursor"], "detect": lambda: True}, }, ): configured = install_platform_configs(tmp_path, target="cursor") assert "Cursor" in configured config_path = tmp_path / ".cursor" / "mcp.json" assert config_path.exists() data = json.loads(config_path.read_text()) assert "code-review-graph" in data["mcpServers"] assert data["mcpServers"]["code-review-graph"]["type"] == "stdio" def test_install_windsurf_config(self, tmp_path): windsurf_dir = tmp_path / ".codeium" / "windsurf" windsurf_dir.mkdir(parents=True) config_path = windsurf_dir / "mcp_config.json" with patch.dict( PLATFORMS, { "windsurf": { **PLATFORMS["windsurf"], "config_path": lambda root: config_path, "detect": lambda: True, }, }, ): configured = install_platform_configs(tmp_path, target="windsurf") assert "Windsurf" in configured data = json.loads(config_path.read_text()) entry = data["mcpServers"]["code-review-graph"] assert "type" not in entry expected_cmd, _ = _detect_serve_command() assert entry["command"] == expected_cmd def test_install_zed_config(self, tmp_path): zed_settings = tmp_path / "zed" / "settings.json" zed_settings.parent.mkdir(parents=True) with patch.dict( PLATFORMS, { "zed": { **PLATFORMS["zed"], "config_path": lambda root: zed_settings, "detect": lambda: True, }, }, ): configured = install_platform_configs(tmp_path, target="zed") assert "Zed" in configured data = json.loads(zed_settings.read_text()) assert "context_servers" in data assert "code-review-graph" in data["context_servers"] def test_install_continue_config(self, tmp_path): continue_dir = tmp_path / ".continue" continue_dir.mkdir() config_path = continue_dir / "config.json" with patch.dict( PLATFORMS, { "continue": { **PLATFORMS["continue"], "config_path": lambda root: config_path, "detect": lambda: True, }, }, ): configured = install_platform_configs(tmp_path, target="continue") assert "Continue" in configured data = json.loads(config_path.read_text()) assert isinstance(data["mcpServers"], list) assert data["mcpServers"][0]["name"] == "code-review-graph" assert data["mcpServers"][0]["type"] == "stdio" def test_install_opencode_config(self, tmp_path): configured = install_platform_configs(tmp_path, target="opencode") assert "OpenCode" in configured config_path = tmp_path / ".opencode.json" data = json.loads(config_path.read_text()) entry = data["mcpServers"]["code-review-graph"] assert entry["type"] == "stdio" assert entry["env"] == [] def test_install_gemini_cli_config(self, tmp_path): gemini_config = tmp_path / ".gemini" / "settings.json" with patch.dict( PLATFORMS, { "gemini-cli": { **PLATFORMS["gemini-cli"], "config_path": lambda root: gemini_config, "detect": lambda: True, }, }, ): configured = install_platform_configs(tmp_path, target="gemini-cli") assert "Gemini CLI" in configured data = json.loads(gemini_config.read_text()) entry = data["mcpServers"]["code-review-graph"] assert "type" not in entry assert entry["args"][-1] == "serve" def test_install_qwen_config(self, tmp_path): """Qwen Code uses ~/.qwen/settings.json with mcpServers (see #83).""" qwen_config = tmp_path / ".qwen" / "settings.json" with patch.dict( PLATFORMS, { "qwen": { **PLATFORMS["qwen"], "config_path": lambda root: qwen_config, "detect": lambda: True, }, }, ): configured = install_platform_configs(tmp_path, target="qwen") assert "Qwen Code" in configured data = json.loads(qwen_config.read_text()) entry = data["mcpServers"]["code-review-graph"] assert entry["type"] == "stdio" assert entry["args"][-1] == "serve" def test_install_qwen_preserves_existing_servers(self, tmp_path): """Adding qwen should merge with, not clobber, existing mcpServers.""" qwen_config = tmp_path / ".qwen" / "settings.json" qwen_config.parent.mkdir(parents=True) qwen_config.write_text( json.dumps({"mcpServers": {"other-server": {"command": "other"}}}), encoding="utf-8", ) with patch.dict( PLATFORMS, { "qwen": { **PLATFORMS["qwen"], "config_path": lambda root: qwen_config, "detect": lambda: True, }, }, ): install_platform_configs(tmp_path, target="qwen") data = json.loads(qwen_config.read_text()) assert "other-server" in data["mcpServers"] assert "code-review-graph" in data["mcpServers"] def test_install_all_detected(self, tmp_path): """Installing 'all' configures auto-detected platforms.""" codex_config = tmp_path / ".codex" / "config.toml" with patch.dict( PLATFORMS, { "codex": { **PLATFORMS["codex"], "config_path": lambda root: codex_config, "detect": lambda: True, }, "claude": {**PLATFORMS["claude"], "detect": lambda: True}, "opencode": {**PLATFORMS["opencode"], "detect": lambda: True}, "cursor": {**PLATFORMS["cursor"], "detect": lambda: False}, "windsurf": {**PLATFORMS["windsurf"], "detect": lambda: False}, "zed": {**PLATFORMS["zed"], "detect": lambda: False}, "continue": {**PLATFORMS["continue"], "detect": lambda: False}, "antigravity": {**PLATFORMS["antigravity"], "detect": lambda: False}, "gemini-cli": {**PLATFORMS["gemini-cli"], "detect": lambda: False}, }, ): configured = install_platform_configs(tmp_path, target="all") assert "Codex" in configured assert "Claude Code" in configured assert "OpenCode" in configured assert codex_config.exists() assert (tmp_path / ".mcp.json").exists() assert (tmp_path / ".opencode.json").exists() def test_merge_existing_servers(self, tmp_path): """Should not overwrite existing MCP servers.""" mcp_path = tmp_path / ".mcp.json" existing = {"mcpServers": {"other-server": {"command": "other"}}} mcp_path.write_text(json.dumps(existing)) install_platform_configs(tmp_path, target="claude") data = json.loads(mcp_path.read_text()) assert "other-server" in data["mcpServers"] assert "code-review-graph" in data["mcpServers"] def test_dry_run_no_write(self, tmp_path): configured = install_platform_configs(tmp_path, target="claude", dry_run=True) assert "Claude Code" in configured assert not (tmp_path / ".mcp.json").exists() def test_already_configured_skips(self, tmp_path): install_platform_configs(tmp_path, target="claude") configured = install_platform_configs(tmp_path, target="claude") assert "Claude Code" in configured def test_continue_array_no_duplicate(self, tmp_path): config_path = tmp_path / ".continue" / "config.json" config_path.parent.mkdir(parents=True) existing = { "mcpServers": [{"name": "code-review-graph", "command": "uvx", "args": ["serve"]}] } config_path.write_text(json.dumps(existing)) with patch.dict( PLATFORMS, { "continue": { **PLATFORMS["continue"], "config_path": lambda root: config_path, "detect": lambda: True, }, }, ): install_platform_configs(tmp_path, target="continue") data = json.loads(config_path.read_text()) assert len(data["mcpServers"]) == 1 def test_install_qoder_config(self, tmp_path): qoder_config = tmp_path / ".qoder" / "mcp.json" with patch.dict( PLATFORMS, { "qoder": { **PLATFORMS["qoder"], "config_path": lambda root: qoder_config, "detect": lambda: True, }, }, ): configured = install_platform_configs(tmp_path, target="qoder") assert "Qoder" in configured data = json.loads(qoder_config.read_text()) assert "mcpServers" in data assert "code-review-graph" in data["mcpServers"] assert data["mcpServers"]["code-review-graph"]["type"] == "stdio" expected_cmd, _ = _detect_serve_command() assert data["mcpServers"]["code-review-graph"]["command"] == expected_cmd class TestGeminiCLIInstall: def test_install_gemini_cli_hooks_creates_settings_and_scripts(self, tmp_path): settings_dir = tmp_path / ".gemini" settings_dir.mkdir(parents=True, exist_ok=True) settings_path = settings_dir / "settings.json" settings_path.write_text(json.dumps({"customSetting": True}) + "\n", encoding="utf-8") out_path = install_gemini_cli_hooks(tmp_path) assert out_path == settings_path assert (settings_dir / "settings.json.bak").exists() data = json.loads(settings_path.read_text(encoding="utf-8")) assert data["customSetting"] is True assert "hooks" in data assert "SessionStart" in data["hooks"] assert "AfterTool" in data["hooks"] session_start = settings_dir / "hooks" / "crg-session-start.sh" update = settings_dir / "hooks" / "crg-update.sh" assert session_start.exists() assert update.exists() assert os.access(session_start, os.X_OK) assert os.access(update, os.X_OK) def test_install_gemini_cli_skills_writes_skill_dirs(self, tmp_path): skills_root = install_gemini_cli_skills(tmp_path) assert skills_root == tmp_path / ".gemini" / "skills" skill_path = skills_root / "explore-codebase" / "SKILL.md" assert skill_path.exists() text = skill_path.read_text(encoding="utf-8") assert text.startswith("---\n") assert "name: explore-codebase" in text assert "description:" in text class TestCursorHooksConfig: """Tests for generate_cursor_hooks_config().""" def test_has_version_1(self): config = generate_cursor_hooks_config() assert config["version"] == 1 def test_has_after_file_edit(self): config = generate_cursor_hooks_config() hooks = config["hooks"]["afterFileEdit"] assert len(hooks) >= 1 assert "crg-update.sh" in hooks[0]["command"] assert hooks[0]["timeout"] == 5 def test_has_session_start(self): config = generate_cursor_hooks_config() hooks = config["hooks"]["sessionStart"] assert len(hooks) >= 1 assert "crg-session-start.sh" in hooks[0]["command"] assert hooks[0]["timeout"] == 5 def test_has_before_shell_execution(self): config = generate_cursor_hooks_config() hooks = config["hooks"]["beforeShellExecution"] assert len(hooks) >= 1 assert "crg-pre-commit.sh" in hooks[0]["command"] assert hooks[0]["timeout"] == 10 assert hooks[0]["matcher"] == "^git\\s+commit" def test_has_all_three_hook_types(self): config = generate_cursor_hooks_config() hook_types = set(config["hooks"].keys()) assert hook_types == {"afterFileEdit", "sessionStart", "beforeShellExecution"} def test_commands_point_to_home_cursor_hooks(self): config = generate_cursor_hooks_config() from pathlib import Path hooks_dir = str(Path.home() / ".cursor" / "hooks") for event, entries in config["hooks"].items(): for entry in entries: assert entry["command"].startswith(hooks_dir), ( f"{event} command does not start with {hooks_dir}" ) class TestCursorHookScripts: """Tests for _cursor_hook_scripts().""" def test_returns_three_scripts(self): scripts = _cursor_hook_scripts() assert set(scripts.keys()) == { "crg-update.sh", "crg-session-start.sh", "crg-pre-commit.sh", } def test_scripts_start_with_shebang(self): scripts = _cursor_hook_scripts() for name, content in scripts.items(): assert content.startswith("#!/usr/bin/env bash"), f"{name} missing shebang line" def test_scripts_exit_zero(self): """Each script must end with exit 0 for graceful failure.""" scripts = _cursor_hook_scripts() for name, content in scripts.items(): assert "exit 0" in content, f"{name} missing 'exit 0'" def test_scripts_consume_stdin(self): """Each script must consume stdin (Cursor protocol).""" scripts = _cursor_hook_scripts() for name, content in scripts.items(): assert "cat > /dev/null" in content, f"{name} missing stdin consumption" def test_update_script_runs_update(self): scripts = _cursor_hook_scripts() assert "code-review-graph update --skip-flows" in scripts["crg-update.sh"] def test_session_start_script_runs_status(self): scripts = _cursor_hook_scripts() assert "code-review-graph status" in scripts["crg-session-start.sh"] def test_pre_commit_script_runs_detect_changes(self): scripts = _cursor_hook_scripts() assert "code-review-graph detect-changes --brief" in scripts["crg-pre-commit.sh"] class TestInstallCursorHooks: """Tests for install_cursor_hooks().""" def test_creates_hooks_json(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): result = install_cursor_hooks() hooks_json = tmp_path / ".cursor" / "hooks.json" assert hooks_json.exists() assert result == hooks_json data = json.loads(hooks_json.read_text()) assert data["version"] == 1 assert "afterFileEdit" in data["hooks"] def test_creates_hook_scripts(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): install_cursor_hooks() hooks_dir = tmp_path / ".cursor" / "hooks" assert (hooks_dir / "crg-update.sh").exists() assert (hooks_dir / "crg-session-start.sh").exists() assert (hooks_dir / "crg-pre-commit.sh").exists() def test_scripts_are_executable(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): install_cursor_hooks() hooks_dir = tmp_path / ".cursor" / "hooks" for script in hooks_dir.iterdir(): mode = script.stat().st_mode assert mode & stat.S_IXUSR, f"{script.name} not executable by owner" assert mode & stat.S_IXGRP, f"{script.name} not executable by group" def test_merges_with_existing_hooks_json(self, tmp_path): cursor_dir = tmp_path / ".cursor" cursor_dir.mkdir(parents=True) existing = { "version": 1, "hooks": { "afterFileEdit": [{"command": "/some/other/hook.sh", "timeout": 3}], "stop": [{"command": "/some/stop-hook.sh", "timeout": 2}], }, } (cursor_dir / "hooks.json").write_text(json.dumps(existing)) with patch("code_review_graph.skills.Path.home", return_value=tmp_path): install_cursor_hooks() data = json.loads((cursor_dir / "hooks.json").read_text()) # Original hook preserved commands = [h["command"] for h in data["hooks"]["afterFileEdit"]] assert "/some/other/hook.sh" in commands # Our hook added assert any("crg-update.sh" in c for c in commands) # Unrelated hook type preserved assert "stop" in data["hooks"] def test_no_duplicate_on_reinstall(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): install_cursor_hooks() install_cursor_hooks() data = json.loads((tmp_path / ".cursor" / "hooks.json").read_text()) # Each event type should have exactly 1 crg hook for event, entries in data["hooks"].items(): crg_hooks = [h for h in entries if "crg-" in h.get("command", "")] assert len(crg_hooks) == 1, f"{event} has {len(crg_hooks)} crg hooks after reinstall" def test_handles_corrupt_existing_json(self, tmp_path): cursor_dir = tmp_path / ".cursor" cursor_dir.mkdir(parents=True) (cursor_dir / "hooks.json").write_text("not valid json{{{") with patch("code_review_graph.skills.Path.home", return_value=tmp_path): result = install_cursor_hooks() assert result.exists() data = json.loads(result.read_text()) assert data["version"] == 1 class TestKiroPlatform: """Tests for Kiro platform support.""" def test_kiro_platform_entry_exists(self): """PLATFORMS dict has a 'kiro' key with correct metadata.""" assert "kiro" in PLATFORMS kiro = PLATFORMS["kiro"] assert kiro["name"] == "Kiro" assert kiro["key"] == "mcpServers" assert kiro["format"] == "object" assert kiro["needs_type"] is True def test_install_kiro_config(self, tmp_path): """install_platform_configs creates .kiro/settings/mcp.json.""" configured = install_platform_configs(tmp_path, target="kiro") assert "Kiro" in configured config_path = tmp_path / ".kiro" / "settings" / "mcp.json" assert config_path.exists() data = json.loads(config_path.read_text()) assert "code-review-graph" in data["mcpServers"] entry = data["mcpServers"]["code-review-graph"] assert entry["type"] == "stdio" def test_install_kiro_preserves_existing_servers(self, tmp_path): """Existing mcpServers entries are preserved when adding code-review-graph.""" config_path = tmp_path / ".kiro" / "settings" / "mcp.json" config_path.parent.mkdir(parents=True) config_path.write_text( json.dumps({"mcpServers": {"other-server": {"command": "other"}}}), encoding="utf-8", ) install_platform_configs(tmp_path, target="kiro") data = json.loads(config_path.read_text()) assert "other-server" in data["mcpServers"] assert "code-review-graph" in data["mcpServers"] def test_install_kiro_no_duplicate(self, tmp_path): """Second install skips when code-review-graph already exists.""" install_platform_configs(tmp_path, target="kiro") config_path = tmp_path / ".kiro" / "settings" / "mcp.json" first_content = config_path.read_text() install_platform_configs(tmp_path, target="kiro") second_content = config_path.read_text() assert first_content == second_content data = json.loads(second_content) assert list(data["mcpServers"].keys()).count("code-review-graph") == 1 def test_kiro_steering_file_written(self, tmp_path): """inject_platform_instructions creates .kiro/steering/code-review-graph.md.""" updated = inject_platform_instructions(tmp_path, target="kiro") assert ".kiro/steering/code-review-graph.md" in updated steering = tmp_path / ".kiro" / "steering" / "code-review-graph.md" assert steering.exists() content = steering.read_text() assert _CLAUDE_MD_SECTION_MARKER in content def test_kiro_steering_idempotent(self, tmp_path): """Running inject twice produces identical content.""" inject_platform_instructions(tmp_path, target="kiro") first = (tmp_path / ".kiro" / "steering" / "code-review-graph.md").read_text() inject_platform_instructions(tmp_path, target="kiro") second = (tmp_path / ".kiro" / "steering" / "code-review-graph.md").read_text() assert first == second def test_kiro_included_in_all_when_detected(self, tmp_path): """install_platform_configs with target='all' includes Kiro when .kiro exists.""" (tmp_path / ".kiro").mkdir() # Mock Path.home() to a dir without .kiro so only workspace detection fires fake_home = tmp_path / "fakehome" fake_home.mkdir() with patch("code_review_graph.skills.Path.home", return_value=fake_home): configured = install_platform_configs(tmp_path, target="all") assert "Kiro" in configured def test_kiro_workspace_detection(self, tmp_path): """Kiro detected when repo_root/.kiro exists even if ~/.kiro does not.""" (tmp_path / ".kiro").mkdir() fake_home = tmp_path / "fakehome" fake_home.mkdir() with patch("code_review_graph.skills.Path.home", return_value=fake_home): configured = install_platform_configs(tmp_path, target="all") assert "Kiro" in configured config_path = tmp_path / ".kiro" / "settings" / "mcp.json" assert config_path.exists() def test_kiro_dry_run(self, tmp_path): """dry_run=True does not create any files.""" configured = install_platform_configs(tmp_path, target="kiro", dry_run=True) assert "Kiro" in configured config_path = tmp_path / ".kiro" / "settings" / "mcp.json" assert not config_path.exists() class TestCopilotPlatform: """Tests for GitHub Copilot platform support.""" def test_copilot_platform_entry_exists(self): """PLATFORMS dict has a 'copilot' key with correct metadata.""" assert "copilot" in PLATFORMS copilot = PLATFORMS["copilot"] assert copilot["name"] == "GitHub Copilot" assert copilot["key"] == "servers" assert copilot["format"] == "object" assert copilot["needs_type"] is True def test_install_copilot_config(self, tmp_path): """install_platform_configs creates .vscode/mcp.json with 'servers' key.""" configured = install_platform_configs(tmp_path, target="copilot") assert "GitHub Copilot" in configured config_path = tmp_path / ".vscode" / "mcp.json" assert config_path.exists() data = json.loads(config_path.read_text()) assert "code-review-graph" in data["servers"] entry = data["servers"]["code-review-graph"] assert entry["type"] == "stdio" assert "serve" in entry["args"] def test_install_copilot_preserves_existing_servers(self, tmp_path): """Existing server entries are preserved when adding code-review-graph.""" config_path = tmp_path / ".vscode" / "mcp.json" config_path.parent.mkdir(parents=True) config_path.write_text( json.dumps({"servers": {"other-server": {"command": "other"}}}), encoding="utf-8", ) install_platform_configs(tmp_path, target="copilot") data = json.loads(config_path.read_text()) assert "other-server" in data["servers"] assert "code-review-graph" in data["servers"] def test_install_copilot_no_duplicate(self, tmp_path): """Second install skips when code-review-graph already exists.""" install_platform_configs(tmp_path, target="copilot") config_path = tmp_path / ".vscode" / "mcp.json" first_content = config_path.read_text() install_platform_configs(tmp_path, target="copilot") second_content = config_path.read_text() assert first_content == second_content data = json.loads(second_content) assert list(data["servers"].keys()).count("code-review-graph") == 1 def test_copilot_instructions_file_written(self, tmp_path): """inject_platform_instructions creates .github/code-review-graph.instruction.md.""" updated = inject_platform_instructions(tmp_path, target="copilot") assert ".github/code-review-graph.instruction.md" in updated instructions = tmp_path / ".github" / "code-review-graph.instruction.md" assert instructions.exists() content = instructions.read_text() assert _CLAUDE_MD_SECTION_MARKER in content def test_copilot_instructions_idempotent(self, tmp_path): """Running inject twice produces identical content.""" inject_platform_instructions(tmp_path, target="copilot") first = (tmp_path / ".github" / "code-review-graph.instruction.md").read_text() inject_platform_instructions(tmp_path, target="copilot") second = (tmp_path / ".github" / "code-review-graph.instruction.md").read_text() assert first == second def test_copilot_dry_run(self, tmp_path): """dry_run=True does not create any files.""" configured = install_platform_configs(tmp_path, target="copilot", dry_run=True) assert "GitHub Copilot" in configured config_path = tmp_path / ".vscode" / "mcp.json" assert not config_path.exists() def test_copilot_writes_only_copilot_instructions(self, tmp_path): """inject_platform_instructions with target='copilot' writes only copilot file.""" updated = inject_platform_instructions(tmp_path, target="copilot") assert updated == [".github/code-review-graph.instruction.md"] assert not (tmp_path / "AGENTS.md").exists() assert not (tmp_path / "GEMINI.md").exists() assert not (tmp_path / ".cursorrules").exists() assert not (tmp_path / ".windsurfrules").exists() assert not (tmp_path / "QODER.md").exists() def test_copilot_included_in_all_when_detected(self, tmp_path): """install_platform_configs with target='all' includes Copilot when ~/.vscode exists.""" fake_home = tmp_path / "fakehome" (fake_home / ".vscode").mkdir(parents=True) with patch("code_review_graph.skills.Path.home", return_value=fake_home): configured = install_platform_configs(tmp_path, target="all") assert "GitHub Copilot" in configured config_path = tmp_path / ".vscode" / "mcp.json" assert config_path.exists() class TestCopilotCLIPlatform: """Tests for GitHub Copilot CLI platform support.""" def test_copilot_cli_platform_entry_exists(self): """PLATFORMS dict has a 'copilot-cli' key with correct metadata.""" assert "copilot-cli" in PLATFORMS copilot_cli = PLATFORMS["copilot-cli"] assert copilot_cli["name"] == "GitHub Copilot CLI" assert copilot_cli["key"] == "servers" assert copilot_cli["format"] == "object" assert copilot_cli["needs_type"] is True def test_install_copilot_cli_config(self, tmp_path): """install_platform_configs creates ~/.copilot/mcp-config.json with 'servers' key.""" fake_home = tmp_path / "fakehome" (fake_home / ".copilot").mkdir(parents=True) config_path = fake_home / ".copilot" / "mcp-config.json" with patch.dict( PLATFORMS, { "copilot-cli": { **PLATFORMS["copilot-cli"], "config_path": lambda root: config_path, "detect": lambda: True, }, }, ): configured = install_platform_configs(tmp_path, target="copilot-cli") assert "GitHub Copilot CLI" in configured assert config_path.exists() data = json.loads(config_path.read_text()) assert "code-review-graph" in data["servers"] entry = data["servers"]["code-review-graph"] assert entry["type"] == "stdio" assert "serve" in entry["args"] def test_install_copilot_cli_preserves_existing_servers(self, tmp_path): """Existing server entries are preserved when adding code-review-graph.""" fake_home = tmp_path / "fakehome" config_path = fake_home / ".copilot" / "mcp-config.json" config_path.parent.mkdir(parents=True) config_path.write_text( json.dumps({"servers": {"other-server": {"command": "other"}}}), encoding="utf-8", ) with patch.dict( PLATFORMS, { "copilot-cli": { **PLATFORMS["copilot-cli"], "config_path": lambda root: config_path, "detect": lambda: True, }, }, ): install_platform_configs(tmp_path, target="copilot-cli") data = json.loads(config_path.read_text()) assert "other-server" in data["servers"] assert "code-review-graph" in data["servers"] def test_copilot_cli_writes_only_copilot_instructions(self, tmp_path): """inject_platform_instructions with target='copilot-cli' writes .github/code-review-graph.instruction.md.""" updated = inject_platform_instructions(tmp_path, target="copilot-cli") assert ".github/code-review-graph.instruction.md" in updated instructions = tmp_path / ".github" / "code-review-graph.instruction.md" assert instructions.exists() content = instructions.read_text() assert _CLAUDE_MD_SECTION_MARKER in content class TestDetectServeCommand: """Tests for _detect_serve_command() and its helpers.""" # ------------------------------------------------------------------ # _in_poetry_project() unit tests # ------------------------------------------------------------------ def test_in_poetry_project_via_poetry_active(self, monkeypatch): """POETRY_ACTIVE=1 signals a poetry shell session.""" monkeypatch.setenv("POETRY_ACTIVE", "1") monkeypatch.delenv("VIRTUAL_ENV", raising=False) assert _in_poetry_project() is True def test_in_poetry_project_via_virtual_env(self, monkeypatch): """VIRTUAL_ENV containing 'pypoetry' signals a poetry run session.""" monkeypatch.delenv("POETRY_ACTIVE", raising=False) monkeypatch.setenv("VIRTUAL_ENV", "/home/user/.cache/pypoetry/virtualenvs/proj-xxx") assert _in_poetry_project() is True def test_in_poetry_project_false_for_plain_venv(self, monkeypatch): """A plain venv (no pypoetry in path) is not treated as poetry.""" monkeypatch.delenv("POETRY_ACTIVE", raising=False) monkeypatch.setenv("VIRTUAL_ENV", "/home/user/myproject/.venv") assert _in_poetry_project() is False def test_in_poetry_project_false_when_nothing_set(self, monkeypatch): """No env vars → not in a poetry project.""" monkeypatch.delenv("POETRY_ACTIVE", raising=False) monkeypatch.delenv("VIRTUAL_ENV", raising=False) assert _in_poetry_project() is False # ------------------------------------------------------------------ # _detect_serve_command() integration tests # ------------------------------------------------------------------ def test_poetry_active_returns_poetry_run(self, monkeypatch): """POETRY_ACTIVE=1 (poetry shell) → 'poetry run' invocation.""" monkeypatch.setenv("POETRY_ACTIVE", "1") monkeypatch.delenv("VIRTUAL_ENV", raising=False) monkeypatch.setattr( "code_review_graph.skills.shutil.which", lambda x: "/usr/bin/poetry" if x == "poetry" else None, ) cmd, args = _detect_serve_command() assert cmd == "poetry" assert args == ["run", "code-review-graph", "serve"] def test_virtual_env_pypoetry_returns_poetry_run(self, monkeypatch): """VIRTUAL_ENV with 'pypoetry' (poetry run) → 'poetry run' invocation.""" monkeypatch.delenv("POETRY_ACTIVE", raising=False) monkeypatch.setenv("VIRTUAL_ENV", "/home/user/.cache/pypoetry/virtualenvs/proj-abc123") monkeypatch.setattr( "code_review_graph.skills.shutil.which", lambda x: "/usr/bin/poetry" if x == "poetry" else None, ) cmd, args = _detect_serve_command() assert cmd == "poetry" assert args == ["run", "code-review-graph", "serve"] def test_poetry_env_without_poetry_on_path_falls_through(self, monkeypatch): """If poetry venv is detected but poetry binary is missing, fall through.""" monkeypatch.setenv("POETRY_ACTIVE", "1") monkeypatch.delenv("VIRTUAL_ENV", raising=False) monkeypatch.delenv("UV_PROJECT_ENVIRONMENT", raising=False) monkeypatch.setattr("code_review_graph.skills._in_uv_project", lambda: False) # poetry not on PATH → should fall through to uvx monkeypatch.setattr( "code_review_graph.skills.shutil.which", lambda x: "/usr/bin/uvx" if x == "uvx" else None, ) cmd, _ = _detect_serve_command() assert cmd == "uvx" def test_uv_project_env_returns_uv_run(self, monkeypatch): """UV_PROJECT_ENVIRONMENT set + uv on PATH → 'uv run' invocation.""" monkeypatch.delenv("POETRY_ACTIVE", raising=False) monkeypatch.delenv("VIRTUAL_ENV", raising=False) monkeypatch.setenv("UV_PROJECT_ENVIRONMENT", "/some/.venv") monkeypatch.setattr( "code_review_graph.skills.shutil.which", lambda x: "/usr/bin/uv" if x == "uv" else None, ) cmd, args = _detect_serve_command() assert cmd == "uv" assert args == ["run", "code-review-graph", "serve"] def test_uv_lock_detection_returns_uv_run(self, monkeypatch, tmp_path): """uv.lock alongside sys.executable → detected as a uv project.""" monkeypatch.delenv("POETRY_ACTIVE", raising=False) monkeypatch.delenv("VIRTUAL_ENV", raising=False) monkeypatch.delenv("UV_PROJECT_ENVIRONMENT", raising=False) venv = tmp_path / ".venv" / "bin" venv.mkdir(parents=True) (tmp_path / "uv.lock").write_text("") fake_python = venv / "python" fake_python.write_text("") monkeypatch.setattr("code_review_graph.skills.sys.executable", str(fake_python)) monkeypatch.setattr( "code_review_graph.skills.shutil.which", lambda x: "/usr/bin/uv" if x == "uv" else None, ) assert _in_uv_project() is True cmd, args = _detect_serve_command() assert cmd == "uv" assert args == ["run", "code-review-graph", "serve"] def test_uvx_fallback(self, monkeypatch): """Not in Poetry/uv but uvx available → use uvx (original behaviour).""" monkeypatch.delenv("POETRY_ACTIVE", raising=False) monkeypatch.delenv("VIRTUAL_ENV", raising=False) monkeypatch.delenv("UV_PROJECT_ENVIRONMENT", raising=False) monkeypatch.setattr("code_review_graph.skills._in_uv_project", lambda: False) monkeypatch.setattr( "code_review_graph.skills.shutil.which", lambda x: "/usr/bin/uvx" if x == "uvx" else None, ) cmd, args = _detect_serve_command() assert cmd == "uvx" assert args == ["code-review-graph", "serve"] def test_sys_executable_fallback(self, monkeypatch): """Nothing else available → fall back to sys.executable -m.""" monkeypatch.delenv("POETRY_ACTIVE", raising=False) monkeypatch.delenv("VIRTUAL_ENV", raising=False) monkeypatch.delenv("UV_PROJECT_ENVIRONMENT", raising=False) monkeypatch.setattr("code_review_graph.skills._in_uv_project", lambda: False) monkeypatch.setattr("code_review_graph.skills.shutil.which", lambda _: None) cmd, args = _detect_serve_command() assert cmd == sys.executable assert args == ["-m", "code_review_graph", "serve"] def test_poetry_takes_priority_over_uv(self, monkeypatch): """Poetry detection wins even when UV_PROJECT_ENVIRONMENT is also set.""" monkeypatch.setenv("POETRY_ACTIVE", "1") monkeypatch.delenv("VIRTUAL_ENV", raising=False) monkeypatch.setenv("UV_PROJECT_ENVIRONMENT", "/some/.venv") monkeypatch.setattr( "code_review_graph.skills.shutil.which", lambda x: "/usr/bin/poetry" if x == "poetry" else None, ) cmd, _ = _detect_serve_command() assert cmd == "poetry" def test_in_uv_project_false_without_lockfile(self, monkeypatch, tmp_path): """_in_uv_project returns False when no uv.lock in ancestor dirs.""" fake_python = tmp_path / "bin" / "python" fake_python.parent.mkdir(parents=True) fake_python.write_text("") monkeypatch.setattr("code_review_graph.skills.sys.executable", str(fake_python)) monkeypatch.setattr("code_review_graph.skills.Path.home", staticmethod(lambda: tmp_path)) assert _in_uv_project() is False class TestOpenCodePluginContent: """Tests for _opencode_plugin_content().""" def test_returns_non_empty_string(self): content = _opencode_plugin_content() assert isinstance(content, str) assert len(content) > 100 def test_has_plugin_type_import(self): content = _opencode_plugin_content() assert "import type" in content assert "@opencode-ai/plugin" in content def test_has_default_export(self): content = _opencode_plugin_content() assert "export default" in content def test_hooks_file_edited_event(self): content = _opencode_plugin_content() assert '"file.edited"' in content assert "code-review-graph update --skip-flows" in content def test_hooks_session_created_event(self): content = _opencode_plugin_content() assert '"session.created"' in content assert "code-review-graph status" in content def test_hooks_tool_execute_before_event(self): content = _opencode_plugin_content() assert '"tool.execute.before"' in content assert "code-review-graph detect-changes --brief" in content def test_has_git_commit_detection(self): """Pre-commit hook should match git commit commands.""" content = _opencode_plugin_content() assert "git" in content assert "commit" in content def test_all_handlers_have_try_catch(self): """Every event handler must use try/catch for graceful failure.""" content = _opencode_plugin_content() # Count the three event registrations and ensure catch blocks assert content.count("} catch") >= 3 class TestInstallOpenCodePlugin: """Tests for install_opencode_plugin().""" def test_creates_plugin_file(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): result = install_opencode_plugin() plugin_path = tmp_path / ".config" / "opencode" / "plugins" / "crg-plugin.ts" assert plugin_path.exists() assert result == plugin_path def test_plugin_file_has_correct_content(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): result = install_opencode_plugin() content = result.read_text(encoding="utf-8") assert "export default" in content assert "file.edited" in content def test_creates_parent_directories(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): install_opencode_plugin() plugins_dir = tmp_path / ".config" / "opencode" / "plugins" assert plugins_dir.is_dir() def test_overwrites_existing_plugin(self, tmp_path): plugins_dir = tmp_path / ".config" / "opencode" / "plugins" plugins_dir.mkdir(parents=True) old_plugin = plugins_dir / "crg-plugin.ts" old_plugin.write_text("// old version") with patch("code_review_graph.skills.Path.home", return_value=tmp_path): install_opencode_plugin() content = old_plugin.read_text() assert "// old version" not in content assert "export default" in content def test_idempotent(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): install_opencode_plugin() result = install_opencode_plugin() content = result.read_text() assert "export default" in content # Only one default export in the file assert content.count("export default") == 1 def test_plugin_is_typescript(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): result = install_opencode_plugin() assert result.suffix == ".ts" def test_preserves_other_plugins(self, tmp_path): plugins_dir = tmp_path / ".config" / "opencode" / "plugins" plugins_dir.mkdir(parents=True) other_plugin = plugins_dir / "other-plugin.ts" other_plugin.write_text("// other plugin") with patch("code_review_graph.skills.Path.home", return_value=tmp_path): install_opencode_plugin() assert other_plugin.exists() assert other_plugin.read_text() == "// other plugin" def test_file_is_utf8(self, tmp_path): with patch("code_review_graph.skills.Path.home", return_value=tmp_path): result = install_opencode_plugin() # Should be readable as UTF-8 without errors content = result.read_text(encoding="utf-8") assert len(content) > 0