1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
"""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 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(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)
|
