loader Public

Watch 0 Fork 0 Star 0
Python · 5761 bytes Raw Blame History
1 """Tests for workflow-oriented tools introduced in Sprint 04."""
2
3 from __future__ import annotations
4
5 import json
6 from pathlib import Path
7
8 import pytest
9
10 from loader.tools.workflow_tools import AskUserQuestionTool, TodoWriteTool
11
12
13 @pytest.mark.asyncio
14 async def test_todo_write_persists_and_returns_previous_state(tmp_path: Path) -> None:
15 tool = TodoWriteTool(tmp_path)
16
17 first = await tool.execute(
18 todos=[
19 {
20 "content": "Create runtime router",
21 "active_form": "Creating runtime router",
22 "status": "in_progress",
23 }
24 ]
25 )
26 second = await tool.execute(
27 todos=[
28 {
29 "content": "Create runtime router",
30 "active_form": "Creating runtime router",
31 "status": "completed",
32 }
33 ]
34 )
35
36 first_payload = json.loads(first.output)
37 second_payload = json.loads(second.output)
38 store_path = tmp_path / ".loader" / "todos" / "active.json"
39
40 assert first.is_error is False
41 assert first_payload["old_todos"] == []
42 assert second_payload["old_todos"] == first_payload["new_todos"]
43 assert json.loads(store_path.read_text()) == []
44
45
46 @pytest.mark.asyncio
47 async def test_todo_write_merges_partial_status_updates_with_existing_scope(
48 tmp_path: Path,
49 ) -> None:
50 tool = TodoWriteTool(tmp_path)
51
52 initial = await tool.execute(
53 todos=[
54 {
55 "content": "Create nginx index",
56 "active_form": "Creating nginx index",
57 "status": "completed",
58 },
59 {
60 "content": "Create chapter files",
61 "active_form": "Creating chapter files",
62 "status": "in_progress",
63 },
64 {
65 "content": "Verify links",
66 "active_form": "Verifying links",
67 "status": "pending",
68 },
69 ]
70 )
71 partial = await tool.execute(
72 todos=[
73 {
74 "content": "Create chapter files",
75 "active_form": "Creating chapter files",
76 "status": "completed",
77 }
78 ]
79 )
80
81 initial_payload = json.loads(initial.output)
82 partial_payload = json.loads(partial.output)
83 assert initial.is_error is False
84 assert partial.is_error is False
85 assert partial_payload["old_todos"] == initial_payload["new_todos"]
86 assert partial_payload["new_todos"] == [
87 {
88 "content": "Create nginx index",
89 "active_form": "Creating nginx index",
90 "status": "completed",
91 },
92 {
93 "content": "Create chapter files",
94 "active_form": "Creating chapter files",
95 "status": "completed",
96 },
97 {
98 "content": "Verify links",
99 "active_form": "Verifying links",
100 "status": "pending",
101 },
102 ]
103
104
105 @pytest.mark.asyncio
106 async def test_todo_write_rejects_invalid_payloads_and_sets_verification_nudge(
107 tmp_path: Path,
108 ) -> None:
109 tool = TodoWriteTool(tmp_path)
110
111 empty = await tool.execute(todos=[])
112 blank = await tool.execute(
113 todos=[
114 {
115 "content": " ",
116 "active_form": "Reviewing plan",
117 "status": "pending",
118 }
119 ]
120 )
121 nudged = await tool.execute(
122 todos=[
123 {
124 "content": "Implement router",
125 "active_form": "Implementing router",
126 "status": "completed",
127 },
128 {
129 "content": "Write tests",
130 "active_form": "Writing tests",
131 "status": "completed",
132 },
133 {
134 "content": "Update docs",
135 "active_form": "Updating docs",
136 "status": "completed",
137 },
138 ]
139 )
140
141 assert empty.is_error is True
142 assert "todos must not be empty" in empty.output
143 assert blank.is_error is True
144 assert "todo content must not be empty" in blank.output
145 assert json.loads(nudged.output)["verification_nudge_needed"] is True
146
147
148 @pytest.mark.asyncio
149 async def test_todo_write_defaults_missing_active_form_to_content(
150 tmp_path: Path,
151 ) -> None:
152 tool = TodoWriteTool(tmp_path)
153
154 result = await tool.execute(
155 todos=[
156 {
157 "content": "Create the nginx chapters content",
158 "status": "completed",
159 }
160 ]
161 )
162
163 payload = json.loads(result.output)
164 assert result.is_error is False
165 assert payload["new_todos"] == [
166 {
167 "content": "Create the nginx chapters content",
168 "active_form": "Create the nginx chapters content",
169 "status": "completed",
170 }
171 ]
172
173
174 @pytest.mark.asyncio
175 async def test_ask_user_question_uses_callback_and_resolves_numbered_options() -> None:
176 tool = AskUserQuestionTool()
177
178 async def answer(question: str, options: list[str] | None) -> str:
179 assert "Which path" in question
180 assert options == ["Plan first", "Execute now"]
181 return "2"
182
183 result = await tool.execute(
184 question="Which path should we take?",
185 options=["Plan first", "Execute now"],
186 user_response_handler=answer,
187 )
188
189 payload = json.loads(result.output)
190 assert result.is_error is False
191 assert payload["answer"] == "Execute now"
192 assert payload["status"] == "answered"
193
194
195 @pytest.mark.asyncio
196 async def test_ask_user_question_requires_callback() -> None:
197 tool = AskUserQuestionTool()
198
199 result = await tool.execute(question="Need an answer?")
200
201 assert result.is_error is True
202 assert "user_response_handler" in result.output