File size: 5,026 Bytes
310260a | 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 | """
Integration Tests for Agent Create Task (User Story 1)
Tests the end-to-end workflow of AI agent creating tasks via chat endpoint.
"""
import pytest
import asyncio
from concurrent.futures import ThreadPoolExecutor
from tests.utils.agent_helpers import (
invoke_agent_with_message,
assert_tool_called,
extract_tool_calls
)
from tests.utils.task_helpers import count_tasks, get_task_by_id
@pytest.mark.integration
@pytest.mark.asyncio
async def test_agent_creates_task_via_chat_endpoint(test_user, test_session, mock_mcp_context):
"""
Test: Agent creates task via chat endpoint
Verifies that AI agent can create a task through natural language interaction.
"""
# Execute
response = await invoke_agent_with_message(
user_id=test_user.id,
message="Add a task to buy groceries"
)
# Assert agent called create_task tool
assert_tool_called(response, "create_task")
# Verify task was created in database
task_count = count_tasks(test_session, test_user.id)
assert task_count == 1
# Verify response contains confirmation
assert response["content"] is not None
assert len(response["content"]) > 0
@pytest.mark.integration
@pytest.mark.asyncio
async def test_agent_creates_task_with_natural_language_remind_me(test_user, test_session):
"""
Test: Agent creates task with natural language "remind me to X"
Verifies that agent recognizes "remind me to" intent and creates task.
"""
# Execute
response = await invoke_agent_with_message(
user_id=test_user.id,
message="Remind me to call the dentist tomorrow"
)
# Assert agent called create_task tool
assert_tool_called(response, "create_task")
# Verify task was created
task_count = count_tasks(test_session, test_user.id)
assert task_count == 1
# Extract tool calls to verify parameters
tool_calls = extract_tool_calls(response)
create_task_call = next((tc for tc in tool_calls if tc["tool"] == "create_task"), None)
assert create_task_call is not None
# Verify tool was called with appropriate title
result = create_task_call.get("result", {})
assert result.get("status") == "success"
@pytest.mark.integration
@pytest.mark.asyncio
async def test_multiple_agents_create_tasks_concurrently_without_conflicts(test_user, test_user2, test_session):
"""
Test: Multiple agents create tasks concurrently without conflicts
Verifies that concurrent task creation by different agents doesn't cause conflicts.
"""
# Define concurrent task creation operations
async def create_task_for_user1():
return await invoke_agent_with_message(
user_id=test_user.id,
message="Add task: User 1 Task A"
)
async def create_task_for_user2():
return await invoke_agent_with_message(
user_id=test_user2.id,
message="Add task: User 2 Task B"
)
# Execute concurrently
results = await asyncio.gather(
create_task_for_user1(),
create_task_for_user2(),
return_exceptions=True
)
# Assert both succeeded
assert len(results) == 2
for result in results:
assert not isinstance(result, Exception)
assert_tool_called(result, "create_task")
# Verify each user has exactly 1 task
user1_count = count_tasks(test_session, test_user.id)
user2_count = count_tasks(test_session, test_user2.id)
assert user1_count == 1
assert user2_count == 1
@pytest.mark.integration
@pytest.mark.asyncio
async def test_agent_creates_multiple_tasks_in_sequence(test_user, test_session):
"""
Test: Agent creates multiple tasks in sequence
Verifies that agent can create multiple tasks in a single conversation.
"""
# Create first task
response1 = await invoke_agent_with_message(
user_id=test_user.id,
message="Add a task to buy milk"
)
assert_tool_called(response1, "create_task")
# Create second task
response2 = await invoke_agent_with_message(
user_id=test_user.id,
message="Also add a task to buy eggs"
)
assert_tool_called(response2, "create_task")
# Verify both tasks were created
task_count = count_tasks(test_session, test_user.id)
assert task_count == 2
@pytest.mark.integration
@pytest.mark.asyncio
async def test_agent_creates_task_with_description(test_user, test_session):
"""
Test: Agent creates task with description from natural language
Verifies that agent can extract both title and description from user message.
"""
# Execute
response = await invoke_agent_with_message(
user_id=test_user.id,
message="Add a task to review the PR. Make sure to check code quality and test coverage."
)
# Assert agent called create_task tool
assert_tool_called(response, "create_task")
# Verify task was created
task_count = count_tasks(test_session, test_user.id)
assert task_count == 1
|