Skip to content
Merged
Show file tree
Hide file tree
Changes from 12 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ dmypy.json
# ===========================
# Node.js (apps/frontend)
# ===========================
node_modules/
node_modules
.npm
.yarn/
.pnp.*
Expand Down
16 changes: 11 additions & 5 deletions apps/backend/runners/github/bot_detection.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@
from datetime import datetime, timedelta
from pathlib import Path

from core.gh_executable import get_gh_executable

try:
from .file_lock import FileLock, atomic_write
except (ImportError, ValueError, SystemError):
Expand Down Expand Up @@ -148,16 +150,20 @@ def _get_bot_username(self) -> str | None:
return None

try:
gh_exec = get_gh_executable()
if not gh_exec:
print(
"[BotDetector] gh CLI not found, cannot identify bot user",
file=sys.stderr,
)
return None

# Use gh api to get authenticated user
# Pass token via environment variable to avoid exposing it in process listings
env = os.environ.copy()
env["GH_TOKEN"] = self.bot_token
result = subprocess.run(
[
"gh",
"api",
"user",
],
[gh_exec, "api", "user"],
capture_output=True,
text=True,
timeout=5,
Expand Down
9 changes: 8 additions & 1 deletion apps/backend/runners/github/gh_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
from pathlib import Path
from typing import Any

from core.gh_executable import get_gh_executable

try:
from .rate_limiter import RateLimiter, RateLimitExceeded
except (ImportError, ValueError, SystemError):
Expand Down Expand Up @@ -129,7 +131,12 @@ async def run(
GHCommandError: If command fails and raise_on_error is True
"""
timeout = timeout or self.default_timeout
cmd = ["gh"] + args
gh_exec = get_gh_executable()
if not gh_exec:
raise GHCommandError(
"GitHub CLI (gh) not found. Install from https://cli.github.com/"
)
cmd = [gh_exec] + args
start_time = asyncio.get_event_loop().time()

# Pre-flight rate limit check
Expand Down
18 changes: 4 additions & 14 deletions apps/backend/runners/github/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,26 +100,16 @@ def print_progress(callback: ProgressCallback) -> None:

def get_config(args) -> GitHubRunnerConfig:
"""Build config from CLI args and environment."""
import shutil
import subprocess

from core.gh_executable import get_gh_executable

token = args.token or os.environ.get("GITHUB_TOKEN", "")
bot_token = args.bot_token or os.environ.get("GITHUB_BOT_TOKEN")
repo = args.repo or os.environ.get("GITHUB_REPO", "")

# Find gh CLI - use shutil.which for cross-platform support
gh_path = shutil.which("gh")
if not gh_path and sys.platform == "win32":
# Fallback: check common Windows installation paths
common_paths = [
r"C:\Program Files\GitHub CLI\gh.exe",
r"C:\Program Files (x86)\GitHub CLI\gh.exe",
os.path.expandvars(r"%LOCALAPPDATA%\Programs\GitHub CLI\gh.exe"),
]
for path in common_paths:
if os.path.exists(path):
gh_path = path
break
# Find gh CLI - use get_gh_executable for cross-platform support
gh_path = get_gh_executable()

if os.environ.get("DEBUG"):
safe_print(f"[DEBUG] gh CLI path: {gh_path}")
Expand Down
119 changes: 119 additions & 0 deletions apps/backend/runners/github/test_bot_detection.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
"""

import json
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path
Expand Down Expand Up @@ -413,5 +414,123 @@ def test_invalid_last_review_time(self, mock_bot_detector):
assert is_cooling is False


class TestGhExecutableDetection:
"""Test gh executable detection in bot_detector._get_bot_username."""

def test_get_bot_username_with_gh_not_found(self, temp_state_dir):
"""Test _get_bot_username when gh CLI is not found."""
with patch("bot_detection.get_gh_executable", return_value=None):
detector = BotDetector(
state_dir=temp_state_dir,
bot_token="fake-token",
review_own_prs=False,
)

# Should not crash, username should be None
assert detector.bot_username is None

def test_get_bot_username_with_detected_gh(self, temp_state_dir):
"""Test _get_bot_username when gh CLI is found."""
mock_gh_path = str(temp_state_dir / "gh")
with patch("bot_detection.get_gh_executable", return_value=mock_gh_path):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
stdout=json.dumps({"login": "test-bot-user"}),
)

detector = BotDetector(
state_dir=temp_state_dir,
bot_token="fake-token",
review_own_prs=False,
)

# Should use the detected gh path
assert detector.bot_username == "test-bot-user"

# Verify subprocess was called with the correct gh path
mock_run.assert_called_once()
called_cmd_list = mock_run.call_args[0][0]
assert called_cmd_list[0] == mock_gh_path
assert called_cmd_list[1:] == ["api", "user"]

def test_get_bot_username_uses_get_gh_executable_return_value(self, temp_state_dir):
"""Test that _get_bot_username uses the path returned by get_gh_executable."""
# Note: GITHUB_CLI_PATH env var is tested by get_gh_executable's own tests
# This test verifies _get_bot_username uses whatever get_gh_executable returns
mock_gh_path = str(temp_state_dir / "gh")

with patch("bot_detection.get_gh_executable", return_value=mock_gh_path):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
stdout=json.dumps({"login": "env-bot-user"}),
)

detector = BotDetector(
state_dir=temp_state_dir,
bot_token="fake-token",
review_own_prs=False,
)

# Verify the command was run with the path from get_gh_executable
assert detector.bot_username == "env-bot-user"

# Verify subprocess was called with the correct path
mock_run.assert_called_once()
called_cmd_list = mock_run.call_args[0][0]
assert called_cmd_list[0] == mock_gh_path
assert called_cmd_list[1:] == ["api", "user"]

def test_get_bot_username_with_api_error(self, temp_state_dir):
"""Test _get_bot_username when gh api command fails."""
mock_gh_path = str(temp_state_dir / "gh")
with patch("bot_detection.get_gh_executable", return_value=mock_gh_path):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=1,
stderr="Authentication failed",
)

detector = BotDetector(
state_dir=temp_state_dir,
bot_token="invalid-token",
review_own_prs=False,
)

# Should handle error gracefully, username should be None
assert detector.bot_username is None

def test_get_bot_username_with_subprocess_timeout(self, temp_state_dir):
"""Test _get_bot_username when subprocess times out."""
mock_gh_path = str(temp_state_dir / "gh")
with patch("bot_detection.get_gh_executable", return_value=mock_gh_path):
with patch(
"subprocess.run", side_effect=subprocess.TimeoutExpired("gh", 5)
):
detector = BotDetector(
state_dir=temp_state_dir,
bot_token="fake-token",
review_own_prs=False,
)

# Should handle timeout gracefully, username should be None
assert detector.bot_username is None

def test_get_bot_username_without_token(self, temp_state_dir):
"""Test _get_bot_username when no bot token is provided."""
with patch("subprocess.run") as mock_run:
detector = BotDetector(
state_dir=temp_state_dir,
bot_token=None,
review_own_prs=False,
)

# Should return None without trying to call gh
assert detector.bot_username is None
# Verify subprocess.run was not called (no gh CLI invocation)
mock_run.assert_not_called()


if __name__ == "__main__":
pytest.main([__file__, "-v"])
47 changes: 47 additions & 0 deletions apps/backend/runners/github/test_gh_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import asyncio
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from gh_client import GHClient, GHCommandError, GHTimeoutError
Expand Down Expand Up @@ -59,5 +60,51 @@ async def test_convenience_methods_timeout_protection(self, client):
await client.issue_list()


class TestGHClientGhExecutableDetection:
"""Test suite for GHClient gh executable detection."""

@pytest.fixture
def client(self, tmp_path):
"""Create a test client."""
return GHClient(
project_dir=tmp_path,
default_timeout=2.0,
max_retries=3,
)

@pytest.mark.asyncio
async def test_run_raises_error_when_gh_not_found(self, client):
"""Test that run() raises GHCommandError when gh is not found."""
with patch("gh_client.get_gh_executable", return_value=None):
with pytest.raises(GHCommandError) as exc_info:
await client.run(["--version"])

assert "not found" in str(exc_info.value)
# Test verifies error message mentions GitHub CLI for user guidance
assert "GitHub CLI" in str(exc_info.value)

@pytest.mark.asyncio
async def test_run_uses_detected_gh_executable(self, client):
"""Test that run() uses the detected gh executable path."""
mock_exec = "/custom/path/to/gh"

with patch("gh_client.get_gh_executable", return_value=mock_exec):
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
# Mock the subprocess to return immediately
mock_proc = MagicMock()
mock_proc.communicate = AsyncMock(
return_value=(b"gh version 2.0.0\n", b"")
)
mock_proc.returncode = 0
mock_subprocess.return_value = mock_proc

await client.run(["--version"])

# Verify the correct gh path was used
mock_subprocess.assert_called_once()
called_cmd = mock_subprocess.call_args[0][0]
assert called_cmd == mock_exec


if __name__ == "__main__":
pytest.main([__file__, "-v"])
Loading