path: root/servers/taskwarrior/tests/test_server.py

"""Tests for TaskWarrior MCP server."""

import json
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from mcp_server_taskwarrior.server import TaskWarriorServer, TaskWarriorError


class TestTaskWarriorServer:
    """Test the TaskWarriorServer class functionality."""
    
    def test_init(self):
        """Test that the server initializes correctly."""
        server = TaskWarriorServer()
        assert server is not None
    
    @pytest.mark.asyncio
    async def test_run_task_command_success(self, taskwarrior_server):
        """Test successful task command execution."""
        with patch('asyncio.create_subprocess_exec') as mock_subprocess:
            # Mock successful subprocess
            mock_process = AsyncMock()
            mock_process.communicate = AsyncMock(return_value=(
                b'{"test": "output"}',
                b'',
            ))
            mock_process.returncode = 0
            mock_subprocess.return_value = mock_process
            
            stdout, stderr, return_code = await taskwarrior_server._run_task_command("export")
            
            assert return_code == 0
            assert stdout == '{"test": "output"}'
            assert stderr == ''
    
    @pytest.mark.asyncio
    async def test_run_task_command_failure(self, taskwarrior_server):
        """Test task command execution failure."""
        with patch('asyncio.create_subprocess_exec') as mock_subprocess:
            # Mock failed subprocess with a generic error (not "not found")
            mock_process = AsyncMock()
            mock_process.communicate = AsyncMock(return_value=(
                b'',
                b'Error: invalid filter expression',
            ))
            mock_process.returncode = 1
            mock_subprocess.return_value = mock_process
            
            with pytest.raises(TaskWarriorError) as exc_info:
                await taskwarrior_server._run_task_command("invalid")
            
            assert "TaskWarrior command failed" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_run_task_command_not_installed(self, taskwarrior_server):
        """Test when TaskWarrior is not installed."""
        with patch('asyncio.create_subprocess_exec', side_effect=FileNotFoundError()):
            with pytest.raises(TaskWarriorError) as exc_info:
                await taskwarrior_server._run_task_command("export")
            
            assert "not installed" in str(exc_info.value).lower()
    
    @pytest.mark.asyncio
    async def test_list_tasks(self, taskwarrior_server):
        """Test listing tasks."""
        mock_tasks = [
            {"uuid": "123", "description": "Test task 1", "status": "pending"},
            {"uuid": "456", "description": "Test task 2", "status": "pending"},
        ]
        
        with patch.object(
            taskwarrior_server, '_run_task_command', 
            new_callable=AsyncMock,
            return_value=(json.dumps(mock_tasks), '', 0)
        ):
            tasks = await taskwarrior_server.list_tasks()
            assert len(tasks) == 2
            assert tasks[0]["uuid"] == "123"
    
    @pytest.mark.asyncio
    async def test_list_tasks_with_filter(self, taskwarrior_server):
        """Test listing tasks with filter."""
        mock_tasks = [{"uuid": "123", "description": "Test", "status": "pending"}]
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(json.dumps(mock_tasks), '', 0)
        ):
            tasks = await taskwarrior_server.list_tasks("project:Home")
            assert len(tasks) == 1
    
    @pytest.mark.asyncio
    async def test_add_task(self, taskwarrior_server):
        """Test adding a task."""
        mock_output = "Created task 1.\n\nUUID: test-uuid-123"
        mock_tasks = [{"uuid": "test-uuid-123", "description": "New task", "status": "pending", "entry": "2024-01-01T00:00:00Z"}]
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ), patch.object(
            taskwarrior_server, 'list_tasks',
            new_callable=AsyncMock,
            return_value=mock_tasks
        ):
            result = await taskwarrior_server.add_task("New task", project="Test")
            assert result["description"] == "New task"
            assert result["uuid"] == "test-uuid-123"
            assert result["project"] == "Test"
    
    @pytest.mark.asyncio
    async def test_add_task_invalid_priority(self, taskwarrior_server):
        """Test adding a task with invalid priority."""
        with pytest.raises(ValueError) as exc_info:
            await taskwarrior_server.add_task("Test", priority="X")
        
        assert "Priority must be H, M, or L" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_add_tasks(self, taskwarrior_server):
        """Test adding multiple tasks."""
        mock_output1 = "Created task 1.\n\nUUID: test-uuid-123"
        mock_output2 = "Created task 2.\n\nUUID: test-uuid-456"
        mock_tasks = [
            {"uuid": "test-uuid-123", "description": "Task 1", "status": "pending", "entry": "2024-01-01T00:00:00Z"},
            {"uuid": "test-uuid-456", "description": "Task 2", "status": "pending", "entry": "2024-01-01T00:00:01Z"},
        ]
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            side_effect=[
                (mock_output1, '', 0),
                (mock_output2, '', 0),
            ]
        ), patch.object(
            taskwarrior_server, 'list_tasks',
            new_callable=AsyncMock,
            side_effect=[
                [mock_tasks[0]],
                [mock_tasks[1]],
            ]
        ):
            tasks_to_add = [
                {"description": "Task 1", "project": "Test"},
                {"description": "Task 2", "priority": "H"},
            ]
            results = await taskwarrior_server.add_tasks(tasks_to_add)
            assert len(results) == 2
            assert results[0]["description"] == "Task 1"
            assert results[0]["uuid"] == "test-uuid-123"
            assert results[1]["description"] == "Task 2"
            assert results[1]["uuid"] == "test-uuid-456"
    
    @pytest.mark.asyncio
    async def test_add_tasks_empty_list(self, taskwarrior_server):
        """Test adding tasks with empty list returns empty results."""
        results = await taskwarrior_server.add_tasks([])
        assert results == []
    
    @pytest.mark.asyncio
    async def test_add_tasks_with_one_task(self, taskwarrior_server):
        """Test adding a single task in list works."""
        mock_output = "Created task 1.\n\nUUID: test-uuid-123"
        mock_tasks = [{"uuid": "test-uuid-123", "description": "Single task", "status": "pending", "entry": "2024-01-01T00:00:00Z"}]
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ), patch.object(
            taskwarrior_server, 'list_tasks',
            new_callable=AsyncMock,
            return_value=mock_tasks
        ):
            tasks_to_add = [{"description": "Single task"}]
            results = await taskwarrior_server.add_tasks(tasks_to_add)
            assert len(results) == 1
            assert results[0]["description"] == "Single task"
            assert results[0]["uuid"] == "test-uuid-123"
    
    @pytest.mark.asyncio
    async def test_done_task(self, taskwarrior_server):
        """Test marking a single task as done."""
        mock_output = "Completed task test-uuid-123."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.done_task(["test-uuid-123"])
            assert result["uuids"] == ["test-uuid-123"]
            assert result["status"] == "completed"
    
    @pytest.mark.asyncio
    async def test_done_task_multiple(self, taskwarrior_server):
        """Test marking multiple tasks as done."""
        mock_output = "Completed task test-uuid-123.\nCompleted task test-uuid-456."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.done_task(["test-uuid-123", "test-uuid-456"])
            assert result["uuids"] == ["test-uuid-123", "test-uuid-456"]
            assert result["status"] == "completed"
            assert len(result["uuids"]) == 2
    
    @pytest.mark.asyncio
    async def test_done_task_empty_list(self, taskwarrior_server):
        """Test that marking tasks with empty list raises ValueError."""
        with pytest.raises(ValueError) as exc_info:
            await taskwarrior_server.done_task([])
        
        assert "At least one UUID is required" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_delete_task_single(self, taskwarrior_server):
        """Test deleting a single task."""
        mock_output = "Deleted task test-uuid-123."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.delete_task(["test-uuid-123"])
            assert result["uuids"] == ["test-uuid-123"]
            assert result["status"] == "deleted"
    
    @pytest.mark.asyncio
    async def test_delete_task_multiple(self, taskwarrior_server):
        """Test deleting multiple tasks."""
        mock_output = "Deleted task test-uuid-123.\nDeleted task test-uuid-456."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.delete_task(["test-uuid-123", "test-uuid-456"])
            assert result["uuids"] == ["test-uuid-123", "test-uuid-456"]
            assert result["status"] == "deleted"
            assert len(result["uuids"]) == 2
    
    @pytest.mark.asyncio
    async def test_delete_task_empty_list(self, taskwarrior_server):
        """Test that deleting tasks with empty list raises ValueError."""
        with pytest.raises(ValueError) as exc_info:
            await taskwarrior_server.delete_task([])
        
        assert "At least one UUID is required" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_manage_context_set(self, taskwarrior_server):
        """Test setting a context."""
        mock_output = "Context 'work' set."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.manage_context("set", "work")
            assert result["action"] == "set"
            assert result["context"] == "work"
    
    @pytest.mark.asyncio
    async def test_manage_context_list(self, taskwarrior_server):
        """Test listing contexts."""
        mock_output = "Name   Type  Definition\nwork   read  project:Work"
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.manage_context("list")
            assert result["action"] == "list"
            assert "contexts" in result
    
    @pytest.mark.asyncio
    async def test_manage_context_invalid_action(self, taskwarrior_server):
        """Test invalid context action."""
        with pytest.raises(ValueError) as exc_info:
            await taskwarrior_server.manage_context("invalid")
        
        assert "Action must be one of" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_manage_context_set_without_name(self, taskwarrior_server):
        """Test setting context without name."""
        with pytest.raises(ValueError) as exc_info:
            await taskwarrior_server.manage_context("set")
        
        assert "Context name is required" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_modify_task_project(self, taskwarrior_server):
        """Test modifying a task's project."""
        mock_output = "Modified 1 task."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.modify_task(
                "uuid:test-uuid-123",
                project="NewProject"
            )
            assert result["filter_expr"] == "uuid:test-uuid-123"
            assert result["modifications"]["project"] == "NewProject"
    
    @pytest.mark.asyncio
    async def test_modify_task_priority(self, taskwarrior_server):
        """Test modifying a task's priority."""
        mock_output = "Modified 1 task."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.modify_task(
                "uuid:test-uuid-123",
                priority="H"
            )
            assert result["modifications"]["priority"] == "H"
    
    @pytest.mark.asyncio
    async def test_modify_task_clear_priority(self, taskwarrior_server):
        """Test clearing a task's priority."""
        mock_output = "Modified 1 task."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.modify_task(
                "uuid:test-uuid-123",
                priority=""
            )
            assert result["modifications"]["priority"] is None
    
    @pytest.mark.asyncio
    async def test_modify_task_due(self, taskwarrior_server):
        """Test modifying a task's due date."""
        mock_output = "Modified 1 task."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.modify_task(
                "uuid:test-uuid-123",
                due="tomorrow"
            )
            assert result["modifications"]["due"] == "tomorrow"
    
    @pytest.mark.asyncio
    async def test_modify_task_description(self, taskwarrior_server):
        """Test modifying a task's description."""
        mock_output = "Modified 1 task."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.modify_task(
                "uuid:test-uuid-123",
                description="New description"
            )
            assert result["modifications"]["description"] == "New description"
    
    @pytest.mark.asyncio
    async def test_modify_task_tags(self, taskwarrior_server):
        """Test modifying a task's tags."""
        mock_output = "Modified 1 task."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.modify_task(
                "uuid:test-uuid-123",
                tags=["+urgent", "-later"]
            )
            assert result["modifications"]["tags"] == ["+urgent", "-later"]
    
    @pytest.mark.asyncio
    async def test_modify_task_multiple_fields(self, taskwarrior_server):
        """Test modifying multiple fields at once."""
        mock_output = "Modified 1 task."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.modify_task(
                "+work",
                project="Work",
                priority="H",
                tags=["+urgent"]
            )
            assert result["filter_expr"] == "+work"
            assert result["modifications"]["project"] == "Work"
            assert result["modifications"]["priority"] == "H"
            assert result["modifications"]["tags"] == ["+urgent"]
    
    @pytest.mark.asyncio
    async def test_modify_task_no_modifications(self, taskwarrior_server):
        """Test that modifying without any parameters raises ValueError."""
        with pytest.raises(ValueError) as exc_info:
            await taskwarrior_server.modify_task("uuid:test-uuid-123")
        
        assert "At least one modification parameter must be provided" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_modify_task_invalid_priority(self, taskwarrior_server):
        """Test that invalid priority raises ValueError."""
        with pytest.raises(ValueError) as exc_info:
            await taskwarrior_server.modify_task(
                "uuid:test-uuid-123",
                priority="X"
            )
        
        assert "Priority must be H, M, or L" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_modify_task_invalid_tag_format(self, taskwarrior_server):
        """Test that tags without +/- prefix raise ValueError."""
        with pytest.raises(ValueError) as exc_info:
            await taskwarrior_server.modify_task(
                "uuid:test-uuid-123",
                tags=["invalidtag"]
            )
        
        assert "Tags must start with + or -" in str(exc_info.value)
    
    @pytest.mark.asyncio
    async def test_modify_task_with_filter_expression(self, taskwarrior_server):
        """Test modifying tasks with complex filter expression."""
        mock_output = "Modified 2 tasks."
        
        with patch.object(
            taskwarrior_server, '_run_task_command',
            new_callable=AsyncMock,
            return_value=(mock_output, '', 0)
        ):
            result = await taskwarrior_server.modify_task(
                "project:Home status:pending",
                priority="M"
            )
            assert result["filter_expr"] == "project:Home status:pending"
            assert result["modifications"]["priority"] == "M"