"""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_done_task(self, taskwarrior_server): """Test marking a 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["uuid"] == "test-uuid-123" assert result["status"] == "completed" @pytest.mark.asyncio async def test_delete_task(self, taskwarrior_server): """Test deleting a 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["uuid"] == "test-uuid-123" assert result["status"] == "deleted" @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)