wehub-resource-sync
c3bf08ac8d
K8s Workspace Integration Tests / k8s-workspace-tests (push) Waiting to run
Pre-commit / run (ubuntu-latest) (push) Waiting to run
Python Unittest Coverage / test (macos-15, 3.11) (push) Waiting to run
Python Unittest Coverage / test (ubuntu-latest, 3.11) (push) Waiting to run
Python Unittest Coverage / test (windows-latest, 3.11) (push) Waiting to run
Web UI / check (push) Waiting to run
961 lines
31 KiB
Python
961 lines
31 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Unit tests for task tools."""
|
|
from unittest.async_case import IsolatedAsyncioTestCase
|
|
|
|
from utils import AnyString
|
|
|
|
from agentscope.state import AgentState
|
|
from agentscope.tool import TaskCreate, TaskGet, TaskList, TaskUpdate
|
|
|
|
|
|
class TestTaskCreate(IsolatedAsyncioTestCase):
|
|
"""Test cases for TaskCreate tool."""
|
|
|
|
async def asyncSetUp(self) -> None:
|
|
"""Set up test fixtures."""
|
|
self.agent_state = AgentState()
|
|
self.task_create = TaskCreate()
|
|
|
|
async def test_create_single_task(self) -> None:
|
|
"""Test creating a single task."""
|
|
result = await self.task_create(
|
|
subject="Test Task 1",
|
|
description="This is a test task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_id}) created successfully: "
|
|
f"Test Task 1",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task was added to agent state using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Test Task 1",
|
|
"description": "This is a test task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
async def test_create_multiple_tasks(self) -> None:
|
|
"""Test creating multiple tasks."""
|
|
# Create first task
|
|
result1 = await self.task_create(
|
|
subject="Task 1",
|
|
description="First task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task1_id = self.agent_state.tasks_context.tasks[0].id
|
|
expected_result1 = {
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task1_id}) created successfully: "
|
|
f"Task 1",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result1.model_dump(mode="json"), expected_result1)
|
|
|
|
# Create second task
|
|
result2 = await self.task_create(
|
|
subject="Task 2",
|
|
description="Second task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task2_id = self.agent_state.tasks_context.tasks[1].id
|
|
expected_result2 = {
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task2_id}) created successfully: "
|
|
f"Task 2",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result2.model_dump(mode="json"), expected_result2)
|
|
|
|
# Create third task with metadata
|
|
result3 = await self.task_create(
|
|
subject="Task 3",
|
|
description="Third task",
|
|
metadata={"priority": "high"},
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task3_id = self.agent_state.tasks_context.tasks[2].id
|
|
expected_result3 = {
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task3_id}) created successfully: "
|
|
f"Task 3",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result3.model_dump(mode="json"), expected_result3)
|
|
|
|
# Check all tasks using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": AnyString(),
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": AnyString(),
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 3",
|
|
"description": "Third task",
|
|
"metadata": {"priority": "high"},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": AnyString(),
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
async def test_create_task_with_metadata(self) -> None:
|
|
"""Test creating a task with metadata."""
|
|
metadata = {"priority": "high", "tags": ["urgent", "bug"]}
|
|
result = await self.task_create(
|
|
subject="Bug Fix",
|
|
description="Fix critical bug",
|
|
metadata=metadata,
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_id}) created successfully: "
|
|
f"Bug Fix",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Bug Fix",
|
|
"description": "Fix critical bug",
|
|
"metadata": {"priority": "high", "tags": ["urgent", "bug"]},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": AnyString(),
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
|
|
class TestTaskList(IsolatedAsyncioTestCase):
|
|
"""Test cases for TaskList tool."""
|
|
|
|
async def asyncSetUp(self) -> None:
|
|
"""Set up test fixtures."""
|
|
self.agent_state = AgentState()
|
|
self.task_list = TaskList()
|
|
self.task_create = TaskCreate()
|
|
|
|
async def test_list_no_tasks(self) -> None:
|
|
"""Test listing when there are no tasks."""
|
|
result = await self.task_list(_agent_state=self.agent_state)
|
|
|
|
self.assertEqual(len(result.content), 1)
|
|
self.assertEqual(result.content[0].text, "No tasks available.")
|
|
|
|
async def test_list_with_tasks(self) -> None:
|
|
"""Test listing when there are tasks."""
|
|
# Create multiple tasks
|
|
await self.task_create(
|
|
subject="Task 1",
|
|
description="First task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
await self.task_create(
|
|
subject="Task 2",
|
|
description="Second task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
await self.task_create(
|
|
subject="Task 3",
|
|
description="Third task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
task1_id = self.agent_state.tasks_context.tasks[0].id
|
|
task2_id = self.agent_state.tasks_context.tasks[1].id
|
|
task3_id = self.agent_state.tasks_context.tasks[2].id
|
|
|
|
# List tasks
|
|
result = await self.task_list(_agent_state=self.agent_state)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"{task1_id} [pending] Task 1\n"
|
|
f"{task2_id} [pending] Task 2\n"
|
|
f"{task3_id} [pending] Task 3",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
|
|
class TestTaskGet(IsolatedAsyncioTestCase):
|
|
"""Test cases for TaskGet tool."""
|
|
|
|
async def asyncSetUp(self) -> None:
|
|
"""Set up test fixtures."""
|
|
self.agent_state = AgentState()
|
|
self.task_get = TaskGet()
|
|
self.task_create = TaskCreate()
|
|
|
|
async def test_get_existing_task(self) -> None:
|
|
"""Test getting an existing task."""
|
|
# Create a task
|
|
await self.task_create(
|
|
subject="Test Task",
|
|
description="This is a test task with details",
|
|
metadata={"priority": "high"},
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
# Get the task
|
|
result = await self.task_get(
|
|
task_id=task_id,
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_id}): Test Task\n"
|
|
f"Status: pending\n"
|
|
f"Description: This is a test task with details\n"
|
|
f"Metadata: {{'priority': 'high'}}",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
async def test_get_nonexistent_task(self) -> None:
|
|
"""Test getting a task that doesn't exist."""
|
|
result = await self.task_get(
|
|
task_id="nonexistent-id",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": "Task not found",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "error",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
|
|
class TestTaskUpdate(IsolatedAsyncioTestCase):
|
|
"""Test cases for TaskUpdate tool."""
|
|
|
|
async def asyncSetUp(self) -> None:
|
|
"""Set up test fixtures."""
|
|
self.agent_state = AgentState()
|
|
self.task_update = TaskUpdate()
|
|
self.task_create = TaskCreate()
|
|
|
|
async def test_update_subject(self) -> None:
|
|
"""Test updating task subject."""
|
|
# Create a task
|
|
await self.task_create(
|
|
subject="Original Subject",
|
|
description="Test description",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
# Update subject
|
|
result = await self.task_update(
|
|
task_id=task_id,
|
|
subject="Updated Subject",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) subject.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Updated Subject",
|
|
"description": "Test description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
async def test_update_description(self) -> None:
|
|
"""Test updating task description."""
|
|
# Create a task
|
|
await self.task_create(
|
|
subject="Test Task",
|
|
description="Original description",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
# Update description
|
|
result = await self.task_update(
|
|
task_id=task_id,
|
|
description="Updated description",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) description.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Updated description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
async def test_update_status(self) -> None:
|
|
"""Test updating task status."""
|
|
# Create a task
|
|
await self.task_create(
|
|
subject="Test Task",
|
|
description="Test description",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
# Update status to in_progress
|
|
result = await self.task_update(
|
|
task_id=task_id,
|
|
status="in_progress",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) status.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "in_progress",
|
|
"id": task_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
# Update status to completed
|
|
result = await self.task_update(
|
|
task_id=task_id,
|
|
status="completed",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) status.\n\n"
|
|
f"Task completed. "
|
|
f"Call TaskList now to find your next available "
|
|
f"task or see if your work unblocked others.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected[0]["state"] = "completed"
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
# Test deleted status - create a new task first
|
|
await self.task_create(
|
|
subject="Task to Delete",
|
|
description="This task will be deleted",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task_to_delete_id = self.agent_state.tasks_context.tasks[1].id
|
|
|
|
# Update status to deleted
|
|
result = await self.task_update(
|
|
task_id=task_to_delete_id,
|
|
status="deleted",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task_to_delete_id}) has been deleted.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check that only the first task remains
|
|
self.assertEqual(len(self.agent_state.tasks_context.tasks), 1)
|
|
self.assertEqual(self.agent_state.tasks_context.tasks[0].id, task_id)
|
|
|
|
async def test_update_owner(self) -> None:
|
|
"""Test updating task owner."""
|
|
# Create a task
|
|
await self.task_create(
|
|
subject="Test Task",
|
|
description="Test description",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
# Update owner
|
|
result = await self.task_update(
|
|
task_id=task_id,
|
|
owner="agent-1",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) owner.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Test Task",
|
|
"description": "Test description",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task_id,
|
|
"owner": "agent-1",
|
|
"blocks": [],
|
|
"blocked_by": [],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
async def test_update_metadata(self) -> None:
|
|
"""Test updating task metadata."""
|
|
# Create a task with initial metadata
|
|
await self.task_create(
|
|
subject="Test Task",
|
|
description="Test description",
|
|
metadata={"priority": "low", "tags": ["test"]},
|
|
_agent_state=self.agent_state,
|
|
)
|
|
task_id = self.agent_state.tasks_context.tasks[0].id
|
|
|
|
# Update metadata
|
|
result = await self.task_update(
|
|
task_id=task_id,
|
|
metadata={"priority": "high", "new_field": "value"},
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task_id}) metadata.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"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": [],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
async def test_update_add_blocks(self) -> None:
|
|
"""Test updating task blocks relationship."""
|
|
# Create two tasks
|
|
await self.task_create(
|
|
subject="Task 1",
|
|
description="First task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
await self.task_create(
|
|
subject="Task 2",
|
|
description="Second task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
task1_id = self.agent_state.tasks_context.tasks[0].id
|
|
task2_id = self.agent_state.tasks_context.tasks[1].id
|
|
|
|
# Update task1 to block task2
|
|
result = await self.task_update(
|
|
task_id=task1_id,
|
|
add_blocks=[task2_id],
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task1_id}) add_blocks.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check tasks using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task1_id,
|
|
"owner": None,
|
|
"blocks": [task2_id],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task2_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [task1_id],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
async def test_update_add_blocked_by(self) -> None:
|
|
"""Test updating task blockedBy relationship."""
|
|
# Create two tasks
|
|
await self.task_create(
|
|
subject="Task 1",
|
|
description="First task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
await self.task_create(
|
|
subject="Task 2",
|
|
description="Second task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
task1_id = self.agent_state.tasks_context.tasks[0].id
|
|
task2_id = self.agent_state.tasks_context.tasks[1].id
|
|
|
|
# Update task2 to be blocked by task1
|
|
result = await self.task_update(
|
|
task_id=task2_id,
|
|
add_blocked_by=[task1_id],
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Update task (id={task2_id}) add_blocked_by.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check tasks using model_dump
|
|
tasks_dump = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
expected = [
|
|
{
|
|
"subject": "Task 1",
|
|
"description": "First task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task1_id,
|
|
"owner": None,
|
|
"blocks": [task2_id],
|
|
"blocked_by": [],
|
|
},
|
|
{
|
|
"subject": "Task 2",
|
|
"description": "Second task",
|
|
"metadata": {},
|
|
"created_at": AnyString(),
|
|
"state": "pending",
|
|
"id": task2_id,
|
|
"owner": None,
|
|
"blocks": [],
|
|
"blocked_by": [task1_id],
|
|
},
|
|
]
|
|
self.assertEqual(tasks_dump, expected)
|
|
|
|
async def test_update_delete_task(self) -> None:
|
|
"""Test deleting a task and removing it from blocks/blocked_by."""
|
|
# Create three tasks with dependencies
|
|
await self.task_create(
|
|
subject="Task 1",
|
|
description="First task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
await self.task_create(
|
|
subject="Task 2",
|
|
description="Second task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
await self.task_create(
|
|
subject="Task 3",
|
|
description="Third task",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
task1_id = self.agent_state.tasks_context.tasks[0].id
|
|
task2_id = self.agent_state.tasks_context.tasks[1].id
|
|
task3_id = self.agent_state.tasks_context.tasks[2].id
|
|
|
|
# Set up dependencies: Task 1 blocks Task 2, Task 2 blocks Task 3
|
|
await self.task_update(
|
|
task_id=task1_id,
|
|
add_blocks=[task2_id],
|
|
_agent_state=self.agent_state,
|
|
)
|
|
await self.task_update(
|
|
task_id=task2_id,
|
|
add_blocks=[task3_id],
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Verify dependencies are set up correctly
|
|
tasks_before = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
self.assertEqual(tasks_before[0]["blocks"], [task2_id])
|
|
self.assertEqual(tasks_before[1]["blocked_by"], [task1_id])
|
|
self.assertEqual(tasks_before[1]["blocks"], [task3_id])
|
|
self.assertEqual(tasks_before[2]["blocked_by"], [task2_id])
|
|
|
|
# Delete task 2 (which is in the middle of the dependency chain)
|
|
result = await self.task_update(
|
|
task_id=task2_id,
|
|
status="deleted",
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": f"Task (id={task2_id}) has been deleted.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "running",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|
|
|
|
# Check task was removed and task2_id was removed from
|
|
# blocks/blocked_by
|
|
self.assertEqual(len(self.agent_state.tasks_context.tasks), 2)
|
|
|
|
tasks_after = [
|
|
task.model_dump() for task in self.agent_state.tasks_context.tasks
|
|
]
|
|
# Task 1 should no longer have task2_id in blocks
|
|
self.assertEqual(tasks_after[0]["id"], task1_id)
|
|
self.assertEqual(tasks_after[0]["blocks"], [])
|
|
|
|
# Task 3 should no longer have task2_id in blocked_by
|
|
self.assertEqual(tasks_after[1]["id"], task3_id)
|
|
self.assertEqual(tasks_after[1]["blocked_by"], [])
|
|
|
|
async def test_update_nonexistent_task(self) -> None:
|
|
"""Test updating a task that doesn't exist."""
|
|
result = await self.task_update(
|
|
task_id="nonexistent-id",
|
|
subject="New Subject",
|
|
description=None,
|
|
_agent_state=self.agent_state,
|
|
)
|
|
|
|
# Check result using model_dump
|
|
result_dump = result.model_dump(mode="json")
|
|
expected_result = {
|
|
"content": [
|
|
{
|
|
"text": "TaskNotFoundError: "
|
|
"The task (id=nonexistent-id) does not exist.",
|
|
"type": "text",
|
|
"id": AnyString(),
|
|
},
|
|
],
|
|
"state": "error",
|
|
"is_last": True,
|
|
"metadata": {},
|
|
"id": AnyString(),
|
|
}
|
|
self.assertDictEqual(result_dump, expected_result)
|