wehub-resource-sync
c3bf08ac8d
K8s Workspace Integration Tests / k8s-workspace-tests (push) Has been cancelled
Pre-commit / run (ubuntu-latest) (push) Has been cancelled
Python Unittest Coverage / test (macos-15, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (ubuntu-latest, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (windows-latest, 3.11) (push) Has been cancelled
Web UI / check (push) Has been cancelled
1188 lines
36 KiB
Python
1188 lines
36 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Unit tests for task tools executed through toolkit."""
|
|
import json
|
|
from typing import Any
|
|
from unittest.async_case import IsolatedAsyncioTestCase
|
|
|
|
from utils import AnyString
|
|
|
|
from agentscope.message import ToolCallBlock
|
|
from agentscope.state import AgentState
|
|
from agentscope.tool import (
|
|
Toolkit,
|
|
ToolResponse,
|
|
TaskCreate,
|
|
TaskGet,
|
|
TaskList,
|
|
TaskUpdate,
|
|
)
|
|
|
|
|
|
class _ToolkitTaskTestBase(IsolatedAsyncioTestCase):
|
|
"""Shared helpers for toolkit task tool tests."""
|
|
|
|
async def asyncSetUp(self) -> None:
|
|
"""Set up shared test fixtures."""
|
|
self.agent_state = AgentState()
|
|
self.toolkit = Toolkit(
|
|
tools=[
|
|
TaskCreate(),
|
|
TaskList(),
|
|
TaskGet(),
|
|
TaskUpdate(),
|
|
],
|
|
)
|
|
|
|
async def _call_tool(
|
|
self,
|
|
name: str,
|
|
tool_input: dict[str, Any],
|
|
tool_call_id: str,
|
|
) -> ToolResponse:
|
|
"""Call a task tool through toolkit and return the final response."""
|
|
response = None
|
|
async for result in self.toolkit.call_tool(
|
|
ToolCallBlock(
|
|
id=tool_call_id,
|
|
name=name,
|
|
input=json.dumps(tool_input),
|
|
),
|
|
self.agent_state,
|
|
):
|
|
if isinstance(result, ToolResponse):
|
|
response = result
|
|
|
|
self.assertIsNotNone(response)
|
|
return response
|
|
|
|
def _dump_tasks(self) -> list[dict[str, Any]]:
|
|
"""Dump all tasks from the agent state for assertions."""
|
|
return [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
|
|
|
|
class TestToolkitTaskCreate(_ToolkitTaskTestBase):
|
|
"""Test cases for TaskCreate through toolkit."""
|
|
|
|
async def test_create_single_task(self) -> None:
|
|
"""Test creating a single task."""
|
|
response = await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Test Task 1",
|
|
"description": "This is a test task",
|
|
},
|
|
tool_call_id="task-create-single",
|
|
)
|
|
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_id}) created successfully: "
|
|
"Test Task 1",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-create-single",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Test Task 1",
|
|
"description": "This is a test task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_create_multiple_tasks(self) -> None:
|
|
"""Test creating multiple tasks."""
|
|
response_1 = await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
},
|
|
tool_call_id="task-create-1",
|
|
)
|
|
task_1_id = self.agent_state.tasks_context.tasks[0].id
|
|
self.assertDictEqual(
|
|
response_1.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_1_id}) created successfully: "
|
|
f"Task 1",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-create-1",
|
|
},
|
|
)
|
|
|
|
response_2 = await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
},
|
|
tool_call_id="task-create-2",
|
|
)
|
|
task_2_id = self.agent_state.tasks_context.tasks[1].id
|
|
self.assertDictEqual(
|
|
response_2.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_2_id}) created successfully: "
|
|
f"Task 2",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-create-2",
|
|
},
|
|
)
|
|
|
|
response_3 = await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 3",
|
|
"description": "Third task",
|
|
"metadata": {"priority": "high"},
|
|
},
|
|
tool_call_id="task-create-3",
|
|
)
|
|
task_3_id = self.agent_state.tasks_context.tasks[2].id
|
|
self.assertDictEqual(
|
|
response_3.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_3_id}) created successfully: "
|
|
f"Task 3",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-create-3",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_1_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_2_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 3",
|
|
"description": "Third task",
|
|
"metadata": {"priority": "high"},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_3_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_create_task_with_metadata(self) -> None:
|
|
"""Test creating a task with metadata."""
|
|
response = await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Bug Fix",
|
|
"description": "Fix critical bug",
|
|
"metadata": {
|
|
"priority": "high",
|
|
"tags": ["urgent", "bug"],
|
|
},
|
|
},
|
|
tool_call_id="task-create-metadata",
|
|
)
|
|
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_id}) created successfully: "
|
|
f"Bug Fix",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-create-metadata",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Bug Fix",
|
|
"description": "Fix critical bug",
|
|
"metadata": {
|
|
"priority": "high",
|
|
"tags": ["urgent", "bug"],
|
|
},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
|
|
class TestToolkitTaskList(_ToolkitTaskTestBase):
|
|
"""Test cases for TaskList through toolkit."""
|
|
|
|
async def test_list_no_tasks(self) -> None:
|
|
"""Test listing when there are no tasks."""
|
|
response = await self._call_tool(
|
|
name="TaskList",
|
|
tool_input={},
|
|
tool_call_id="task-list-empty",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": "No tasks available.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-list-empty",
|
|
},
|
|
)
|
|
self.assertEqual(self._dump_tasks(), [])
|
|
|
|
async def test_list_with_tasks(self) -> None:
|
|
"""Test listing when there are tasks."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
},
|
|
tool_call_id="task-list-create-1",
|
|
)
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
},
|
|
tool_call_id="task-list-create-2",
|
|
)
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 3",
|
|
"description": "Third task",
|
|
},
|
|
tool_call_id="task-list-create-3",
|
|
)
|
|
|
|
task_1_id = self.agent_state.tasks_context.tasks[0].id
|
|
task_2_id = self.agent_state.tasks_context.tasks[1].id
|
|
task_3_id = self.agent_state.tasks_context.tasks[2].id
|
|
|
|
response = await self._call_tool(
|
|
name="TaskList",
|
|
tool_input={},
|
|
tool_call_id="task-list-with-tasks",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"{task_1_id} [pending] Task 1\n"
|
|
f"{task_2_id} [pending] Task 2\n"
|
|
f"{task_3_id} [pending] Task 3",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-list-with-tasks",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_1_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_2_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 3",
|
|
"description": "Third task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_3_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
|
|
class TestToolkitTaskGet(_ToolkitTaskTestBase):
|
|
"""Test cases for TaskGet through toolkit."""
|
|
|
|
async def test_get_existing_task(self) -> None:
|
|
"""Test getting an existing task."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Test Task",
|
|
"description": "This is a test task with details",
|
|
"metadata": {"priority": "high"},
|
|
},
|
|
tool_call_id="task-get-create",
|
|
)
|
|
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
response = await self._call_tool(
|
|
name="TaskGet",
|
|
tool_input={"task_id": task_id},
|
|
tool_call_id="task-get-existing",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(mode="json"),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_id}): Test Task\n"
|
|
"Status: pending\n"
|
|
"Description: This is a test task with details\n"
|
|
"Metadata: {'priority': 'high'}",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-get-existing",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "This is a test task with details",
|
|
"metadata": {"priority": "high"},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_get_nonexistent_task(self) -> None:
|
|
"""Test getting a task that does not exist."""
|
|
response = await self._call_tool(
|
|
name="TaskGet",
|
|
tool_input={"task_id": "nonexistent-id"},
|
|
tool_call_id="task-get-missing",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": "Task not found",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "error",
|
|
"metadata": {},
|
|
"id": "task-get-missing",
|
|
},
|
|
)
|
|
self.assertEqual(self._dump_tasks(), [])
|
|
|
|
|
|
class TestToolkitTaskUpdate(_ToolkitTaskTestBase):
|
|
"""Test cases for TaskUpdate through toolkit."""
|
|
|
|
async def test_update_subject(self) -> None:
|
|
"""Test updating task subject."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Original Subject",
|
|
"description": "Test description",
|
|
},
|
|
tool_call_id="task-update-subject-create",
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_id,
|
|
"subject": "Updated Subject",
|
|
},
|
|
tool_call_id="task-update-subject",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) subject.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-subject",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Updated Subject",
|
|
"description": "Test description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_update_description(self) -> None:
|
|
"""Test updating task description."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Test Task",
|
|
"description": "Original description",
|
|
},
|
|
tool_call_id="task-update-description-create",
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_id,
|
|
"description": "Updated description",
|
|
},
|
|
tool_call_id="task-update-description",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) description.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-description",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Updated description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_update_status(self) -> None:
|
|
"""Test updating task status."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
},
|
|
tool_call_id="task-update-status-create",
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_id,
|
|
"status": "in_progress",
|
|
},
|
|
tool_call_id="task-update-status-in-progress",
|
|
)
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) status.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-status-in-progress",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "in_progress",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_id,
|
|
"status": "completed",
|
|
},
|
|
tool_call_id="task-update-status-completed",
|
|
)
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) status.\n\n"
|
|
"Task completed. Call TaskList now to find your "
|
|
"next available task or see if your work "
|
|
"unblocked others.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-status-completed",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "completed",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task to Delete",
|
|
"description": "This task will be deleted",
|
|
},
|
|
tool_call_id="task-update-status-create-delete",
|
|
)
|
|
task_to_delete_id = self.agent_state.tasks_context.tasks[1].id
|
|
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_to_delete_id,
|
|
"status": "deleted",
|
|
},
|
|
tool_call_id="task-update-status-deleted",
|
|
)
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_to_delete_id}) has been "
|
|
f"deleted.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-status-deleted",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "completed",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_update_owner(self) -> None:
|
|
"""Test updating task owner."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
},
|
|
tool_call_id="task-update-owner-create",
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_id,
|
|
"owner": "agent-1",
|
|
},
|
|
tool_call_id="task-update-owner",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) owner.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-owner",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": "agent-1",
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_update_metadata(self) -> None:
|
|
"""Test updating task metadata."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
"metadata": {
|
|
"priority": "low",
|
|
"tags": ["test"],
|
|
},
|
|
},
|
|
tool_call_id="task-update-metadata-create",
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_id,
|
|
"metadata": {
|
|
"priority": "high",
|
|
"new_field": "value",
|
|
},
|
|
},
|
|
tool_call_id="task-update-metadata",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) metadata.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-metadata",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
"metadata": {
|
|
"priority": "high",
|
|
"tags": ["test"],
|
|
"new_field": "value",
|
|
},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_update_add_blocks(self) -> None:
|
|
"""Test updating task blocks relationship."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
},
|
|
tool_call_id="task-update-add-blocks-create-1",
|
|
)
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
},
|
|
tool_call_id="task-update-add-blocks-create-2",
|
|
)
|
|
|
|
task_1_id = self.agent_state.tasks_context.tasks[0].id
|
|
task_2_id = self.agent_state.tasks_context.tasks[1].id
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_1_id,
|
|
"add_blocks": [task_2_id],
|
|
},
|
|
tool_call_id="task-update-add-blocks",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_1_id}) add_blocks.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-add-blocks",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_1_id,
|
|
"owner": None,
|
|
"blocks": [task_2_id],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_2_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [task_1_id],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_update_add_blocked_by(self) -> None:
|
|
"""Test updating task blocked_by relationship."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
},
|
|
tool_call_id="task-update-add-blocked-by-create-1",
|
|
)
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
},
|
|
tool_call_id="task-update-add-blocked-by-create-2",
|
|
)
|
|
|
|
task_1_id = self.agent_state.tasks_context.tasks[0].id
|
|
task_2_id = self.agent_state.tasks_context.tasks[1].id
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_2_id,
|
|
"add_blocked_by": [task_1_id],
|
|
},
|
|
tool_call_id="task-update-add-blocked-by",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_2_id}) "
|
|
f"add_blocked_by.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-add-blocked-by",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_1_id,
|
|
"owner": None,
|
|
"blocks": [task_2_id],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_2_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [task_1_id],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_update_delete_task(self) -> None:
|
|
"""Test deleting a task and cleaning dependency relations."""
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
},
|
|
tool_call_id="task-update-delete-create-1",
|
|
)
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
},
|
|
tool_call_id="task-update-delete-create-2",
|
|
)
|
|
await self._call_tool(
|
|
name="TaskCreate",
|
|
tool_input={
|
|
"subject": "Task 3",
|
|
"description": "Third task",
|
|
},
|
|
tool_call_id="task-update-delete-create-3",
|
|
)
|
|
|
|
task_1_id = self.agent_state.tasks_context.tasks[0].id
|
|
task_2_id = self.agent_state.tasks_context.tasks[1].id
|
|
task_3_id = self.agent_state.tasks_context.tasks[2].id
|
|
|
|
await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_1_id,
|
|
"add_blocks": [task_2_id],
|
|
},
|
|
tool_call_id="task-update-delete-link-1",
|
|
)
|
|
await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_2_id,
|
|
"add_blocks": [task_3_id],
|
|
},
|
|
tool_call_id="task-update-delete-link-2",
|
|
)
|
|
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_1_id,
|
|
"owner": None,
|
|
"blocks": [task_2_id],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_2_id,
|
|
"owner": None,
|
|
"blocks": [task_3_id],
|
|
"blocked_by": [task_1_id],
|
|
},
|
|
{
|
|
"subject": "Task 3",
|
|
"description": "Third task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_3_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [task_2_id],
|
|
},
|
|
],
|
|
)
|
|
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": task_2_id,
|
|
"status": "deleted",
|
|
},
|
|
tool_call_id="task-update-delete",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_2_id}) has been deleted.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "success",
|
|
"metadata": {},
|
|
"id": "task-update-delete",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
self._dump_tasks(),
|
|
[
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_1_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 3",
|
|
"description": "Third task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_3_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
],
|
|
)
|
|
|
|
async def test_update_nonexistent_task(self) -> None:
|
|
"""Test updating a task that does not exist."""
|
|
response = await self._call_tool(
|
|
name="TaskUpdate",
|
|
tool_input={
|
|
"task_id": "nonexistent-id",
|
|
"subject": "New Subject",
|
|
"description": None,
|
|
},
|
|
tool_call_id="task-update-missing",
|
|
)
|
|
|
|
self.assertDictEqual(
|
|
response.model_dump(),
|
|
{
|
|
"content": [
|
|
{
|
|
"text": "TaskNotFoundError: "
|
|
"The task (id=nonexistent-id) does not exist.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "error",
|
|
"metadata": {},
|
|
"id": "task-update-missing",
|
|
},
|
|
)
|
|
self.assertEqual(self._dump_tasks(), [])
|