66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
|
|
from web.task_progress import progress_waiting_for_user, project_progress_payloads
|
|
|
|
|
|
def _call(args: dict) -> dict:
|
|
return {
|
|
"role": "assistant",
|
|
"tool_calls": [{
|
|
"function": {
|
|
"name": "task_progress",
|
|
"arguments": json.dumps(args, ensure_ascii=False),
|
|
},
|
|
}],
|
|
}
|
|
|
|
|
|
class TaskProgressProjectionTests(unittest.TestCase):
|
|
def test_latest_full_snapshot_replaces_prior_snapshot(self) -> None:
|
|
steps, seen = project_progress_payloads([
|
|
_call({"steps": [
|
|
{"id": "old", "title": "旧计划", "status": "in_progress"},
|
|
]}),
|
|
_call({"steps": [
|
|
{"id": "s1", "title": "分析", "status": "completed"},
|
|
{"id": "s2", "title": "实现", "status": "in_progress"},
|
|
]}),
|
|
])
|
|
|
|
self.assertTrue(seen)
|
|
self.assertEqual([step["id"] for step in steps], ["s1", "s2"])
|
|
|
|
def test_legacy_updates_remain_replayable(self) -> None:
|
|
steps, seen = project_progress_payloads([
|
|
_call({"action": "set_plan", "steps": [
|
|
{"id": "s1", "title": "分析", "status": "in_progress"},
|
|
{"id": "s2", "title": "实现", "status": "pending"},
|
|
]}),
|
|
_call({"action": "update_step", "step": {
|
|
"id": "s2", "status": "completed",
|
|
}}),
|
|
])
|
|
|
|
self.assertTrue(seen)
|
|
self.assertEqual([step["status"] for step in steps], ["completed", "completed"])
|
|
|
|
def test_ask_user_marks_the_run_as_waiting(self) -> None:
|
|
payloads = [
|
|
_call({"steps": [
|
|
{"id": "s1", "title": "确认方案", "status": "in_progress"},
|
|
]}),
|
|
{
|
|
"role": "assistant",
|
|
"tool_calls": [{"function": {"name": "ask_user", "arguments": "{}"}}],
|
|
},
|
|
]
|
|
|
|
self.assertTrue(progress_waiting_for_user(payloads))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|