"""
Tests for GitHub Issues integration with amicus-mcp.
These tests validate the GitHub Issues workflow including:
- Creating GitHub issues
- Listing GitHub issues
- Importing issues to local task context
- Syncing task status back to GitHub issues
"""
import pytest
import os
import time
from pathlib import Path
from unittest.mock import Mock, patch, MagicMock
from amicus.server import (
create_github_issue as _create_github_issue,
list_github_issues as _list_github_issues,
import_github_issue_to_task as _import_github_issue_to_task,
sync_task_to_github_issue as _sync_task_to_github_issue,
GITHUB_AVAILABLE
)
from amicus.core import read_with_lock, write_with_lock, get_state_file
# Access the underlying functions from FunctionTool wrappers
create_github_issue = _create_github_issue.fn
list_github_issues = _list_github_issues.fn
import_github_issue_to_task = _import_github_issue_to_task.fn
sync_task_to_github_issue = _sync_task_to_github_issue.fn
class TestGitHubIssuesIntegration:
"""Test suite for GitHub Issues integration."""
def test_github_library_availability(self):
"""Test that PyGithub library is available."""
assert GITHUB_AVAILABLE, "PyGithub library should be installed for GitHub integration"
@patch.dict(os.environ, {}, clear=True)
def test_create_github_issue_no_token(self):
"""Test that create_github_issue fails gracefully without GITHUB_TOKEN."""
result = create_github_issue(
repo_owner="test",
repo_name="repo",
title="Test Issue",
body="Test body"
)
assert "GITHUB_TOKEN" in result
assert "ERROR" in result
@patch.dict(os.environ, {}, clear=True)
def test_list_github_issues_no_token(self):
"""Test that list_github_issues fails gracefully without GITHUB_TOKEN."""
result = list_github_issues(
repo_owner="test",
repo_name="repo"
)
assert "GITHUB_TOKEN" in result
assert "ERROR" in result
@patch.dict(os.environ, {}, clear=True)
def test_import_github_issue_no_token(self):
"""Test that import_github_issue_to_task fails gracefully without GITHUB_TOKEN."""
result = import_github_issue_to_task(
repo_owner="test",
repo_name="repo",
issue_number=1
)
assert "GITHUB_TOKEN" in result
assert "ERROR" in result
@patch.dict(os.environ, {"GITHUB_TOKEN": "fake_token"})
@patch("amicus.server.Github")
def test_create_github_issue_success(self, mock_github_class):
"""Test successful GitHub issue creation."""
# Setup mocks
mock_github = Mock()
mock_repo = Mock()
mock_issue = Mock()
mock_issue.html_url = "https://github.com/test/repo/issues/1"
mock_issue.number = 1
mock_repo.create_issue.return_value = mock_issue
mock_github.get_repo.return_value = mock_repo
mock_github_class.return_value = mock_github
# Test
result = create_github_issue(
repo_owner="test",
repo_name="repo",
title="Test Issue",
body="Test body",
labels=["bug", "enhancement"]
)
# Verify
assert "created successfully" in result
assert "https://github.com/test/repo/issues/1" in result
mock_repo.create_issue.assert_called_once_with(
title="Test Issue",
body="Test body",
labels=["bug", "enhancement"]
)
@patch.dict(os.environ, {"GITHUB_TOKEN": "fake_token"})
@patch("amicus.server.Github")
def test_create_github_issue_with_assignment(self, mock_github_class):
"""Test GitHub issue creation with user assignment."""
# Setup mocks
mock_github = Mock()
mock_repo = Mock()
mock_issue = Mock()
mock_user = Mock()
mock_issue.html_url = "https://github.com/test/repo/issues/1"
mock_issue.number = 1
mock_repo.create_issue.return_value = mock_issue
mock_github.get_repo.return_value = mock_repo
mock_github.get_user.return_value = mock_user
mock_github_class.return_value = mock_github
# Test
result = create_github_issue(
repo_owner="test",
repo_name="repo",
title="Test Issue",
body="Test body",
assign_to_user=True
)
# Verify
assert "created successfully" in result
mock_issue.add_to_assignees.assert_called_once_with(mock_user)
@patch.dict(os.environ, {"GITHUB_TOKEN": "fake_token"})
@patch("amicus.server.Github")
def test_list_github_issues_success(self, mock_github_class):
"""Test successful listing of GitHub issues."""
# Setup mocks
mock_github = Mock()
mock_repo = Mock()
mock_issue1 = Mock()
mock_issue1.number = 1
mock_issue1.title = "Issue 1"
mock_issue1.state = "open"
mock_issue1.html_url = "https://github.com/test/repo/issues/1"
mock_issue1.labels = []
mock_issue1.assignee = None
mock_issue1.created_at = "2024-01-01"
mock_issue1.updated_at = "2024-01-02"
mock_issue1.pull_request = None
mock_issues = Mock()
mock_issues.__iter__ = Mock(return_value=iter([mock_issue1]))
mock_issues.totalCount = 1
mock_repo.get_issues.return_value = mock_issues
mock_github.get_repo.return_value = mock_repo
mock_github_class.return_value = mock_github
# Test
result = list_github_issues(
repo_owner="test",
repo_name="repo",
state="open",
limit=10
)
# Verify
assert "Issue 1" in result
assert "#1" in result
assert "Showing 1 of 1" in result
@patch.dict(os.environ, {"GITHUB_TOKEN": "fake_token"})
@patch("amicus.server.Github")
def test_import_github_issue_to_task(self, mock_github_class, temp_context_dir):
"""Test importing a GitHub issue to local task context."""
# Setup mocks
mock_github = Mock()
mock_repo = Mock()
mock_issue = Mock()
mock_issue.number = 3
mock_issue.title = "Test Issue for Import"
mock_issue.html_url = "https://github.com/test/repo/issues/3"
mock_repo.get_issue.return_value = mock_issue
mock_github.get_repo.return_value = mock_repo
mock_github_class.return_value = mock_github
# Initialize state
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test
result = import_github_issue_to_task(
repo_owner="test",
repo_name="repo",
issue_number=3,
priority="high"
)
# Verify
assert "imported as local task" in result
assert "Test Issue for Import" in result
# Check state was updated
state_data = read_with_lock(state_file)
assert len(state_data["next_steps"]) == 1
task = state_data["next_steps"][0]
assert task["github_issue"] == 3
assert task["github_repo"] == "test/repo"
assert task["priority"] == "high"
assert task["status"] == "pending"
assert "GitHub Issue #3" in task["task"]
@patch.dict(os.environ, {"GITHUB_TOKEN": "fake_token"})
@patch("amicus.server.Github")
def test_import_duplicate_github_issue(self, mock_github_class, temp_context_dir):
"""Test that importing the same GitHub issue twice is prevented."""
# Setup mocks
mock_github = Mock()
mock_repo = Mock()
mock_issue = Mock()
mock_issue.number = 3
mock_issue.title = "Test Issue"
mock_issue.html_url = "https://github.com/test/repo/issues/3"
mock_repo.get_issue.return_value = mock_issue
mock_github.get_repo.return_value = mock_repo
mock_github_class.return_value = mock_github
# Initialize state with existing task
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [{
"task": "GitHub Issue #3: Test Issue",
"status": "pending",
"github_issue": 3,
"github_repo": "test/repo"
}],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test
result = import_github_issue_to_task(
repo_owner="test",
repo_name="repo",
issue_number=3
)
# Verify
assert "already imported" in result
@patch.dict(os.environ, {"GITHUB_TOKEN": "fake_token"})
@patch("amicus.server.Github")
def test_sync_task_to_github_issue_with_comment(self, mock_github_class, temp_context_dir):
"""Test syncing a task to GitHub with a comment."""
# Setup mocks
mock_github = Mock()
mock_repo = Mock()
mock_issue = Mock()
mock_issue.number = 3
mock_issue.state = "open"
mock_issue.html_url = "https://github.com/test/repo/issues/3"
mock_repo.get_issue.return_value = mock_issue
mock_github.get_repo.return_value = mock_repo
mock_github_class.return_value = mock_github
# Initialize state with linked task
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [{
"task": "GitHub Issue #3: Test Issue",
"status": "in_progress",
"github_issue": 3,
"github_repo": "test/repo",
"assigned_to": "test-agent"
}],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test
result = sync_task_to_github_issue(
task_index=1,
comment="Work in progress on this issue",
close_issue=False
)
# Verify
assert "updated successfully" in result
mock_issue.create_comment.assert_called_once_with("Work in progress on this issue")
mock_issue.edit.assert_not_called()
@patch.dict(os.environ, {"GITHUB_TOKEN": "fake_token"})
@patch("amicus.server.Github")
def test_sync_task_to_github_issue_and_close(self, mock_github_class, temp_context_dir):
"""Test syncing a task to GitHub and closing the issue."""
# Setup mocks
mock_github = Mock()
mock_repo = Mock()
mock_issue = Mock()
mock_issue.number = 3
mock_issue.state = "open"
mock_issue.html_url = "https://github.com/test/repo/issues/3"
mock_repo.get_issue.return_value = mock_issue
mock_github.get_repo.return_value = mock_repo
mock_github_class.return_value = mock_github
# Initialize state with linked task
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [{
"task": "GitHub Issue #3: Test Issue",
"status": "completed",
"github_issue": 3,
"github_repo": "test/repo",
"completed_by": "test-agent"
}],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test
result = sync_task_to_github_issue(
task_index=1,
comment="Task completed successfully",
close_issue=True
)
# Verify
assert "closed successfully" in result
mock_issue.create_comment.assert_called_once_with("Task completed successfully")
mock_issue.edit.assert_called_once_with(state="closed")
def test_sync_task_without_github_link(self, temp_context_dir):
"""Test that syncing a task without GitHub link fails gracefully."""
# Initialize state with non-linked task
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [{
"task": "Regular task without GitHub link",
"status": "completed"
}],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test
result = sync_task_to_github_issue(
task_index=1,
comment="Test comment"
)
# Verify
assert "not linked to a GitHub issue" in result
assert "ERROR" in result
class TestGitHubIssuesWorkflowIntegration:
"""Integration tests for the complete GitHub Issues workflow."""
@patch.dict(os.environ, {"GITHUB_TOKEN": "fake_token"})
@patch("amicus.server.Github")
def test_complete_workflow(self, mock_github_class, temp_context_dir):
"""Test the complete workflow: create -> import -> work -> sync -> close."""
# Setup mocks
mock_github = Mock()
mock_repo = Mock()
# Mock for create_issue
mock_new_issue = Mock()
mock_new_issue.html_url = "https://github.com/test/repo/issues/5"
mock_new_issue.number = 5
mock_new_issue.title = "Integration Test Issue"
mock_repo.create_issue.return_value = mock_new_issue
# Mock for get_issue (used in import)
mock_issue = Mock()
mock_issue.number = 5
mock_issue.title = "Integration Test Issue"
mock_issue.html_url = "https://github.com/test/repo/issues/5"
mock_issue.state = "open"
mock_repo.get_issue.return_value = mock_issue
mock_github.get_repo.return_value = mock_repo
mock_github_class.return_value = mock_github
# Initialize state
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Step 1: Create GitHub issue
create_result = create_github_issue(
repo_owner="test",
repo_name="repo",
title="Integration Test Issue",
body="This is a test issue for integration testing"
)
assert "created successfully" in create_result
# Step 2: Import issue to local task context
import_result = import_github_issue_to_task(
repo_owner="test",
repo_name="repo",
issue_number=5,
priority="high"
)
assert "imported as local task" in import_result
# Verify task was created
state_data = read_with_lock(state_file)
assert len(state_data["next_steps"]) == 1
task = state_data["next_steps"][0]
assert task["github_issue"] == 5
assert task["status"] == "pending"
# Step 3: Simulate working on the task (mark as in_progress)
task["status"] = "in_progress"
task["assigned_to"] = "test-agent"
write_with_lock(state_file, state_data)
# Step 4: Sync progress to GitHub
sync_result1 = sync_task_to_github_issue(
task_index=1,
comment="Working on this issue",
close_issue=False
)
assert "updated successfully" in sync_result1
mock_issue.create_comment.assert_called_with("Working on this issue")
# Step 5: Complete the task and close the issue
task["status"] = "completed"
task["completed_by"] = "test-agent"
write_with_lock(state_file, state_data)
sync_result2 = sync_task_to_github_issue(
task_index=1,
comment="Task completed successfully",
close_issue=True
)
assert "closed successfully" in sync_result2
mock_issue.edit.assert_called_with(state="closed")
if __name__ == "__main__":
pytest.main([__file__, "-v"])
