48c93adade
New Features:
- /add-project slash command for adding projects without restart
- Clones git repository with shallow clone (--depth 1)
- Updates config.yaml atomically with rollback on failure
- Live config reload (no bot restart needed)
- Administrator permission required
- Comprehensive validation and error handling
Implementation:
- config.py: add_project() and save() methods
- commands.py: add_project command (193 lines)
- 12 new tests covering all scenarios
- Full documentation in COMMANDS_USAGE.md
Test Results: 30/30 passing (100%)
Usage:
/add-project
project_name: my-project
git_url: https://git.manticorum.com/cal/my-project.git
model: sonnet
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
828 lines
31 KiB
Python
828 lines
31 KiB
Python
"""
|
|
Tests for Discord slash commands.
|
|
|
|
Tests cover /reset, /status, and /model commands with:
|
|
- Success scenarios
|
|
- Error handling
|
|
- Permission checks
|
|
- Edge cases
|
|
"""
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import discord
|
|
import pytest
|
|
from discord import app_commands
|
|
from discord.ext import commands
|
|
|
|
from claude_coordinator.commands import ClaudeCommands, ResetConfirmView
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_bot():
|
|
"""Create a mock Discord bot."""
|
|
bot = MagicMock(spec=commands.Bot)
|
|
bot.session_manager = AsyncMock()
|
|
bot.config = MagicMock()
|
|
return bot
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_interaction():
|
|
"""Create a mock Discord interaction."""
|
|
interaction = MagicMock(spec=discord.Interaction)
|
|
interaction.response = AsyncMock()
|
|
interaction.followup = AsyncMock()
|
|
interaction.channel = MagicMock()
|
|
interaction.channel.id = 123456789
|
|
interaction.channel.mention = "#test-channel"
|
|
interaction.user = MagicMock()
|
|
interaction.user.id = 987654321
|
|
return interaction
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_project():
|
|
"""Create a mock project configuration."""
|
|
project = MagicMock()
|
|
project.name = "test-project"
|
|
project.model = "claude-sonnet-4-5"
|
|
return project
|
|
|
|
|
|
@pytest.fixture
|
|
def commands_cog(mock_bot):
|
|
"""Create ClaudeCommands cog instance."""
|
|
return ClaudeCommands(mock_bot)
|
|
|
|
|
|
class TestResetCommand:
|
|
"""Tests for /reset command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_success(self, commands_cog, mock_interaction, mock_project):
|
|
"""Test successful reset command with existing session."""
|
|
# Setup mocks
|
|
commands_cog.config.get_project_by_channel.return_value = mock_project
|
|
session_data = {
|
|
'channel_id': '123456789',
|
|
'session_id': 'test-session-id',
|
|
'project_name': 'test-project',
|
|
'message_count': 42
|
|
}
|
|
commands_cog.session_manager.get_session.return_value = session_data
|
|
|
|
# Execute command - use callback to bypass decorator
|
|
await commands_cog.reset_command.callback(commands_cog, mock_interaction, channel=None)
|
|
|
|
# Verify confirmation message was sent
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
call_args = mock_interaction.response.send_message.call_args
|
|
assert "Reset Session Confirmation" in call_args[0][0] or "Reset Session Confirmation" in str(call_args[1])
|
|
|
|
# Verify view was attached
|
|
assert 'view' in call_args[1]
|
|
assert isinstance(call_args[1]['view'], ResetConfirmView)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_no_session(self, commands_cog, mock_interaction, mock_project):
|
|
"""Test reset command when no session exists."""
|
|
# Setup mocks
|
|
commands_cog.config.get_project_by_channel.return_value = mock_project
|
|
commands_cog.session_manager.get_session.return_value = None
|
|
|
|
# Execute command
|
|
await commands_cog.reset_command.callback(commands_cog, mock_interaction, channel=None)
|
|
|
|
# Verify informational message
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
call_args = mock_interaction.response.send_message.call_args
|
|
content = call_args[0][0] if call_args[0] else call_args[1].get('content', '')
|
|
assert "No active session" in content or call_args[1].get('ephemeral') is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_unconfigured_channel(self, commands_cog, mock_interaction):
|
|
"""Test reset command on unconfigured channel."""
|
|
# Setup mocks
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Execute command
|
|
await commands_cog.reset_command.callback(commands_cog, mock_interaction, channel=None)
|
|
|
|
# Verify error message
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_with_target_channel(self, commands_cog, mock_interaction, mock_project):
|
|
"""Test reset command with explicit target channel."""
|
|
# Setup target channel
|
|
target_channel = MagicMock()
|
|
target_channel.id = 999888777
|
|
target_channel.mention = "#target-channel"
|
|
|
|
# Setup mocks
|
|
commands_cog.config.get_project_by_channel.return_value = mock_project
|
|
session_data = {'channel_id': '999888777', 'project_name': 'test', 'message_count': 5}
|
|
commands_cog.session_manager.get_session.return_value = session_data
|
|
|
|
# Execute command
|
|
await commands_cog.reset_command.callback(commands_cog, mock_interaction, channel=target_channel)
|
|
|
|
# Verify target channel was used
|
|
commands_cog.session_manager.get_session.assert_called_once_with('999888777')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_error_handling(self, commands_cog, mock_interaction, mock_project):
|
|
"""Test reset command error handling."""
|
|
# Setup mock to raise exception
|
|
commands_cog.config.get_project_by_channel.return_value = mock_project
|
|
commands_cog.session_manager.get_session.side_effect = Exception("Database error")
|
|
|
|
# Execute command
|
|
await commands_cog.reset_command.callback(commands_cog, mock_interaction, channel=None)
|
|
|
|
# Verify error message
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
|
|
|
|
class TestResetConfirmView:
|
|
"""Tests for ResetConfirmView interaction."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirm_button_success(self, mock_bot):
|
|
"""Test confirmation button successfully resets session."""
|
|
# Setup
|
|
session_manager = AsyncMock()
|
|
session_manager.reset_session.return_value = True
|
|
|
|
channel = MagicMock()
|
|
channel.mention = "#test-channel"
|
|
|
|
session = {'project_name': 'test-project'}
|
|
|
|
view = ResetConfirmView(session_manager, '123456789', channel, session)
|
|
|
|
# Mock interaction
|
|
interaction = MagicMock()
|
|
interaction.response = AsyncMock()
|
|
interaction.user = MagicMock()
|
|
interaction.user.id = 123
|
|
|
|
# Execute confirm button
|
|
await view.confirm_button.callback(interaction)
|
|
|
|
# Verify reset was called
|
|
session_manager.reset_session.assert_called_once_with('123456789')
|
|
|
|
# Verify success message
|
|
interaction.response.edit_message.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_button(self, mock_bot):
|
|
"""Test cancel button dismisses confirmation."""
|
|
# Setup
|
|
view = ResetConfirmView(
|
|
AsyncMock(),
|
|
'123456789',
|
|
MagicMock(),
|
|
{}
|
|
)
|
|
|
|
interaction = MagicMock()
|
|
interaction.response = AsyncMock()
|
|
|
|
# Execute cancel button
|
|
await view.cancel_button.callback(interaction)
|
|
|
|
# Verify cancellation message
|
|
interaction.response.edit_message.assert_called_once()
|
|
|
|
|
|
class TestStatusCommand:
|
|
"""Tests for /status command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_with_sessions(self, commands_cog, mock_interaction):
|
|
"""Test status command with active sessions."""
|
|
# Setup mock data
|
|
sessions = [
|
|
{
|
|
'channel_id': '123456789',
|
|
'project_name': 'project-1',
|
|
'message_count': 42,
|
|
'last_active': datetime.now().isoformat()
|
|
},
|
|
{
|
|
'channel_id': '987654321',
|
|
'project_name': 'project-2',
|
|
'message_count': 15,
|
|
'last_active': datetime.now().isoformat()
|
|
}
|
|
]
|
|
|
|
stats = {
|
|
'total_sessions': 2,
|
|
'total_messages': 57
|
|
}
|
|
|
|
commands_cog.session_manager.list_sessions.return_value = sessions
|
|
commands_cog.session_manager.get_stats.return_value = stats
|
|
|
|
# Mock get_channel
|
|
mock_channel = MagicMock()
|
|
mock_channel.mention = "#test-channel"
|
|
commands_cog.bot.get_channel.return_value = mock_channel
|
|
|
|
# Execute command
|
|
await commands_cog.status_command.callback(commands_cog, mock_interaction)
|
|
|
|
# Verify defer was called
|
|
mock_interaction.response.defer.assert_called_once_with(ephemeral=True)
|
|
|
|
# Verify embed was sent
|
|
mock_interaction.followup.send.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_no_sessions(self, commands_cog, mock_interaction):
|
|
"""Test status command with no active sessions."""
|
|
# Setup empty data
|
|
commands_cog.session_manager.list_sessions.return_value = []
|
|
commands_cog.session_manager.get_stats.return_value = {'total_sessions': 0}
|
|
|
|
# Execute command
|
|
await commands_cog.status_command.callback(commands_cog, mock_interaction)
|
|
|
|
# Verify message
|
|
mock_interaction.followup.send.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_error_handling(self, commands_cog, mock_interaction):
|
|
"""Test status command error handling."""
|
|
# Setup exception
|
|
commands_cog.session_manager.list_sessions.side_effect = Exception("DB error")
|
|
|
|
# Execute command
|
|
await commands_cog.status_command.callback(commands_cog, mock_interaction)
|
|
|
|
# Verify error message
|
|
mock_interaction.followup.send.assert_called_once()
|
|
|
|
|
|
class TestModelCommand:
|
|
"""Tests for /model command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_switch_success(self, commands_cog, mock_interaction, mock_project):
|
|
"""Test successful model switch."""
|
|
# Setup mocks
|
|
commands_cog.config.get_project_by_channel.return_value = mock_project
|
|
commands_cog.config.save = MagicMock()
|
|
|
|
# Execute command
|
|
await commands_cog.model_command.callback(commands_cog, mock_interaction, "opus")
|
|
|
|
# Verify model was updated
|
|
assert mock_project.model == "claude-opus-4-6"
|
|
|
|
# Verify config was saved
|
|
commands_cog.config.save.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_unconfigured_channel(self, commands_cog, mock_interaction):
|
|
"""Test model command on unconfigured channel."""
|
|
# Setup
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Execute command
|
|
await commands_cog.model_command.callback(commands_cog, mock_interaction, "sonnet")
|
|
|
|
# Verify error message
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_all_choices(self, commands_cog, mock_interaction, mock_project):
|
|
"""Test all model choices work correctly."""
|
|
commands_cog.config.get_project_by_channel.return_value = mock_project
|
|
commands_cog.config.save = MagicMock()
|
|
|
|
test_cases = [
|
|
("sonnet", "claude-sonnet-4-5"),
|
|
("opus", "claude-opus-4-6"),
|
|
("haiku", "claude-3-5-haiku")
|
|
]
|
|
|
|
for model_name, expected_model in test_cases:
|
|
# Execute
|
|
await commands_cog.model_command.callback(commands_cog, mock_interaction, model_name)
|
|
|
|
# Verify
|
|
assert mock_project.model == expected_model
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_error_handling(self, commands_cog, mock_interaction, mock_project):
|
|
"""Test model command error handling."""
|
|
# Setup exception during save
|
|
commands_cog.config.get_project_by_channel.return_value = mock_project
|
|
commands_cog.config.save.side_effect = Exception("Save error")
|
|
|
|
# Execute command
|
|
await commands_cog.model_command.callback(commands_cog, mock_interaction, "sonnet")
|
|
|
|
# Verify error message was sent
|
|
mock_interaction.response.send_message.assert_called()
|
|
|
|
|
|
class TestPermissions:
|
|
"""Test permission handling."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_requires_permissions(self, commands_cog, mock_interaction):
|
|
"""Test that reset command checks permissions."""
|
|
# The @app_commands.checks.has_permissions decorator is applied
|
|
# We verify it exists on the command
|
|
assert hasattr(commands_cog.reset_command, 'checks')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_error_handler(self, commands_cog, mock_interaction):
|
|
"""Test permission error handler."""
|
|
# Create permission error
|
|
error = app_commands.MissingPermissions(['manage_messages'])
|
|
|
|
# Call error handler
|
|
await commands_cog.reset_error(mock_interaction, error)
|
|
|
|
# Verify error message was sent
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
|
|
|
|
class TestCogSetup:
|
|
"""Test cog setup and initialization."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cog_initialization(self, mock_bot):
|
|
"""Test ClaudeCommands cog initializes correctly."""
|
|
cog = ClaudeCommands(mock_bot)
|
|
|
|
assert cog.bot == mock_bot
|
|
assert cog.session_manager == mock_bot.session_manager
|
|
assert cog.config == mock_bot.config
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_setup_function(self, mock_bot):
|
|
"""Test setup function adds cog to bot."""
|
|
from claude_coordinator.commands import setup
|
|
|
|
mock_bot.add_cog = AsyncMock()
|
|
|
|
await setup(mock_bot)
|
|
|
|
# Verify add_cog was called
|
|
mock_bot.add_cog.assert_called_once()
|
|
call_args = mock_bot.add_cog.call_args
|
|
assert isinstance(call_args[0][0], ClaudeCommands)
|
|
"""
|
|
Tests for /add-project command.
|
|
|
|
Tests cover project addition workflow with:
|
|
- Success scenarios
|
|
- Validation (project name, duplicate channel)
|
|
- Git clone handling (success, failure, timeout)
|
|
- Configuration management
|
|
- Rollback on errors
|
|
- Permission checks
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch, call
|
|
|
|
import discord
|
|
import pytest
|
|
from discord import app_commands
|
|
|
|
from claude_coordinator.commands import ClaudeCommands
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_bot():
|
|
"""Create a mock Discord bot."""
|
|
bot = MagicMock()
|
|
bot.session_manager = AsyncMock()
|
|
bot.config = MagicMock()
|
|
return bot
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_interaction():
|
|
"""Create a mock Discord interaction."""
|
|
interaction = MagicMock(spec=discord.Interaction)
|
|
interaction.response = AsyncMock()
|
|
interaction.followup = AsyncMock()
|
|
interaction.channel = MagicMock()
|
|
interaction.channel.id = 123456789
|
|
interaction.channel.mention = "#test-channel"
|
|
interaction.user = MagicMock()
|
|
interaction.user.id = 987654321
|
|
return interaction
|
|
|
|
|
|
@pytest.fixture
|
|
def commands_cog(mock_bot):
|
|
"""Create ClaudeCommands cog instance."""
|
|
return ClaudeCommands(mock_bot)
|
|
|
|
|
|
class TestAddProjectCommand:
|
|
"""Tests for /add-project command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_success(self, commands_cog, mock_interaction):
|
|
"""Test successful project addition with all steps completing."""
|
|
# Setup mocks
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Mock git clone subprocess
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate.return_value = (b'', b'')
|
|
|
|
# Mock file system operations
|
|
with patch('asyncio.create_subprocess_exec', return_value=mock_process), \
|
|
patch('asyncio.wait_for', return_value=(b'', b'')), \
|
|
patch('os.path.exists') as mock_exists, \
|
|
patch('os.path.join', return_value='/opt/projects/test-project/.git'):
|
|
|
|
# First exists() check: directory doesn't exist (False)
|
|
# Second exists() check: .git directory exists (True)
|
|
mock_exists.side_effect = [False, True]
|
|
|
|
# Mock config operations
|
|
commands_cog.config.add_project = MagicMock()
|
|
commands_cog.config.save = MagicMock()
|
|
commands_cog.config.load = MagicMock()
|
|
|
|
# Execute command
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"test-project",
|
|
"https://git.example.com/test.git",
|
|
None,
|
|
"sonnet"
|
|
)
|
|
|
|
# Verify defer was called
|
|
mock_interaction.response.defer.assert_called_once_with(ephemeral=True)
|
|
|
|
# Verify config was updated
|
|
commands_cog.config.add_project.assert_called_once()
|
|
commands_cog.config.save.assert_called_once()
|
|
commands_cog.config.load.assert_called_once()
|
|
|
|
# Verify success message
|
|
mock_interaction.followup.send.assert_called_once()
|
|
call_args = mock_interaction.followup.send.call_args
|
|
assert 'embed' in call_args[1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_duplicate_channel(self, commands_cog, mock_interaction):
|
|
"""Test add-project rejects if channel already configured."""
|
|
# Setup: channel already has a project
|
|
mock_project = MagicMock()
|
|
commands_cog.config.get_project_by_channel.return_value = mock_project
|
|
|
|
# Execute command
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"test-project",
|
|
"https://git.example.com/test.git",
|
|
None,
|
|
"sonnet"
|
|
)
|
|
|
|
# Verify error message
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
call_args = mock_interaction.response.send_message.call_args
|
|
assert "already configured" in call_args[0][0]
|
|
assert call_args[1]['ephemeral'] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_invalid_name(self, commands_cog, mock_interaction):
|
|
"""Test add-project rejects invalid project names."""
|
|
# Setup
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
invalid_names = [
|
|
"Test-Project", # Uppercase
|
|
"test project", # Space
|
|
"test@project", # Special char
|
|
"test.project", # Dot
|
|
]
|
|
|
|
for invalid_name in invalid_names:
|
|
mock_interaction.response.reset_mock()
|
|
|
|
# Execute command
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
invalid_name,
|
|
"https://git.example.com/test.git",
|
|
None,
|
|
"sonnet"
|
|
)
|
|
|
|
# Verify error message
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
call_args = mock_interaction.response.send_message.call_args
|
|
assert "lowercase alphanumeric" in call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_directory_exists(self, commands_cog, mock_interaction):
|
|
"""Test add-project fails if directory already exists."""
|
|
# Setup
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
with patch('os.path.exists', return_value=True):
|
|
# Execute command
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"test-project",
|
|
"https://git.example.com/test.git",
|
|
None,
|
|
"sonnet"
|
|
)
|
|
|
|
# Verify defer was called
|
|
mock_interaction.response.defer.assert_called_once()
|
|
|
|
# Verify error message
|
|
mock_interaction.followup.send.assert_called_once()
|
|
call_args = mock_interaction.followup.send.call_args
|
|
assert "already exists" in call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_git_clone_failure(self, commands_cog, mock_interaction):
|
|
"""Test add-project handles git clone failure gracefully."""
|
|
# Setup
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Mock git clone subprocess with failure
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 1
|
|
mock_process.communicate.return_value = (b'', b'fatal: repository not found')
|
|
|
|
with patch('asyncio.create_subprocess_exec', return_value=mock_process), \
|
|
patch('asyncio.wait_for', return_value=(b'', b'fatal: repository not found')), \
|
|
patch('os.path.exists', return_value=False):
|
|
|
|
# Execute command
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"test-project",
|
|
"https://git.example.com/test.git",
|
|
None,
|
|
"sonnet"
|
|
)
|
|
|
|
# Verify error message
|
|
mock_interaction.followup.send.assert_called_once()
|
|
call_args = mock_interaction.followup.send.call_args
|
|
assert "Git clone failed" in call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_git_timeout(self, commands_cog, mock_interaction):
|
|
"""Test add-project handles git clone timeout."""
|
|
# Setup
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Mock git clone subprocess
|
|
mock_process = AsyncMock()
|
|
mock_process.kill = MagicMock()
|
|
mock_process.wait = AsyncMock()
|
|
|
|
with patch('asyncio.create_subprocess_exec', return_value=mock_process), \
|
|
patch('asyncio.wait_for', side_effect=asyncio.TimeoutError), \
|
|
patch('os.path.exists', return_value=False):
|
|
|
|
# Execute command
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"test-project",
|
|
"https://git.example.com/test.git",
|
|
None,
|
|
"sonnet"
|
|
)
|
|
|
|
# Verify process was killed
|
|
mock_process.kill.assert_called_once()
|
|
|
|
# Verify error message
|
|
mock_interaction.followup.send.assert_called_once()
|
|
call_args = mock_interaction.followup.send.call_args
|
|
assert "timed out" in call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_not_git_repo(self, commands_cog, mock_interaction):
|
|
"""Test add-project validates cloned directory is a git repo."""
|
|
# Setup
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Mock git clone subprocess (success)
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate.return_value = (b'', b'')
|
|
|
|
with patch('asyncio.create_subprocess_exec', return_value=mock_process), \
|
|
patch('asyncio.wait_for', return_value=(b'', b'')), \
|
|
patch('os.path.exists') as mock_exists, \
|
|
patch('os.path.join', return_value='/opt/projects/test-project/.git'), \
|
|
patch('shutil.rmtree') as mock_rmtree:
|
|
|
|
# First exists(): directory doesn't exist (False)
|
|
# Second exists(): .git directory doesn't exist (False)
|
|
mock_exists.side_effect = [False, False]
|
|
|
|
# Execute command
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"test-project",
|
|
"https://git.example.com/test.git",
|
|
None,
|
|
"sonnet"
|
|
)
|
|
|
|
# Verify cleanup was attempted
|
|
mock_rmtree.assert_called_once()
|
|
|
|
# Verify error message
|
|
mock_interaction.followup.send.assert_called_once()
|
|
call_args = mock_interaction.followup.send.call_args
|
|
assert "doesn't appear to be a git repository" in call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_config_rollback(self, commands_cog, mock_interaction):
|
|
"""Test add-project rolls back on config save failure."""
|
|
# Setup
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Mock git clone subprocess (success)
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate.return_value = (b'', b'')
|
|
|
|
with patch('asyncio.create_subprocess_exec', return_value=mock_process), \
|
|
patch('asyncio.wait_for', return_value=(b'', b'')), \
|
|
patch('os.path.exists') as mock_exists, \
|
|
patch('os.path.join', return_value='/opt/projects/test-project/.git'), \
|
|
patch('shutil.rmtree') as mock_rmtree:
|
|
|
|
# exists() checks: directory doesn't exist, .git exists
|
|
mock_exists.side_effect = [False, True]
|
|
|
|
# Mock config operations with save() failing
|
|
commands_cog.config.add_project = MagicMock()
|
|
commands_cog.config.save = MagicMock(side_effect=IOError("Disk full"))
|
|
|
|
# Execute command
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"test-project",
|
|
"https://git.example.com/test.git",
|
|
None,
|
|
"sonnet"
|
|
)
|
|
|
|
# Verify rollback was performed
|
|
mock_rmtree.assert_called_once_with('/opt/projects/test-project')
|
|
|
|
# Verify error message mentions rollback
|
|
mock_interaction.followup.send.assert_called_once()
|
|
call_args = mock_interaction.followup.send.call_args
|
|
assert "Rolled back" in call_args[0][0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_permission_check(self, commands_cog, mock_interaction):
|
|
"""Test add-project command has administrator permission requirement."""
|
|
# Verify the command has the permission decorator
|
|
assert hasattr(commands_cog.add_project_command, 'checks')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_error_handler(self, commands_cog, mock_interaction):
|
|
"""Test add-project permission error handler."""
|
|
# Create permission error
|
|
error = app_commands.MissingPermissions(['administrator'])
|
|
|
|
# Call error handler
|
|
await commands_cog.add_project_error(mock_interaction, error)
|
|
|
|
# Verify error message was sent
|
|
mock_interaction.response.send_message.assert_called_once()
|
|
call_args = mock_interaction.response.send_message.call_args
|
|
assert "Administrator" in call_args[0][0]
|
|
assert call_args[1]['ephemeral'] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_with_target_channel(self, commands_cog, mock_interaction):
|
|
"""Test add-project with explicit target channel parameter."""
|
|
# Setup target channel
|
|
target_channel = MagicMock()
|
|
target_channel.id = 999888777
|
|
target_channel.mention = "#target-channel"
|
|
|
|
# Setup mocks
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Mock git clone subprocess
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate.return_value = (b'', b'')
|
|
|
|
with patch('asyncio.create_subprocess_exec', return_value=mock_process), \
|
|
patch('asyncio.wait_for', return_value=(b'', b'')), \
|
|
patch('os.path.exists') as mock_exists, \
|
|
patch('os.path.join', return_value='/opt/projects/test-project/.git'):
|
|
|
|
mock_exists.side_effect = [False, True]
|
|
|
|
# Mock config operations
|
|
commands_cog.config.add_project = MagicMock()
|
|
commands_cog.config.save = MagicMock()
|
|
commands_cog.config.load = MagicMock()
|
|
|
|
# Execute command with target channel
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"test-project",
|
|
"https://git.example.com/test.git",
|
|
target_channel, # Explicit channel
|
|
"opus"
|
|
)
|
|
|
|
# Verify target channel was used
|
|
add_project_call = commands_cog.config.add_project.call_args
|
|
assert add_project_call[0][0]['channel_id'] == '999888777'
|
|
assert add_project_call[0][0]['model'] == 'opus'
|
|
|
|
|
|
class TestAddProjectIntegration:
|
|
"""Integration tests for add-project configuration."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_config_structure(self, commands_cog, mock_interaction):
|
|
"""Test add-project creates correct config structure."""
|
|
# Setup
|
|
commands_cog.config.get_project_by_channel.return_value = None
|
|
|
|
# Capture the add_project call
|
|
captured_config = None
|
|
def capture_config(config):
|
|
nonlocal captured_config
|
|
captured_config = config
|
|
|
|
commands_cog.config.add_project = MagicMock(side_effect=capture_config)
|
|
commands_cog.config.save = MagicMock()
|
|
commands_cog.config.load = MagicMock()
|
|
|
|
# Mock git operations
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate.return_value = (b'', b'')
|
|
|
|
with patch('asyncio.create_subprocess_exec', return_value=mock_process), \
|
|
patch('asyncio.wait_for', return_value=(b'', b'')), \
|
|
patch('os.path.exists') as mock_exists, \
|
|
patch('os.path.join', return_value='/opt/projects/my-project/.git'):
|
|
|
|
mock_exists.side_effect = [False, True]
|
|
|
|
# Execute
|
|
await commands_cog.add_project_command.callback(
|
|
commands_cog,
|
|
mock_interaction,
|
|
"my-project",
|
|
"https://git.example.com/my-project.git",
|
|
None,
|
|
"haiku"
|
|
)
|
|
|
|
# Verify config structure
|
|
assert captured_config is not None
|
|
assert captured_config['name'] == 'my-project'
|
|
assert captured_config['channel_id'] == '123456789'
|
|
assert captured_config['project_dir'] == '/opt/projects/my-project'
|
|
assert captured_config['model'] == 'haiku'
|
|
assert 'Bash' in captured_config['allowed_tools']
|
|
assert 'Read' in captured_config['allowed_tools']
|