File size: 3,743 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 | """
Concurrency Tests
Tests concurrent access and tool invocations:
- Multiple agents create tasks concurrently
- Multiple agents update same task concurrently
- Concurrent chat requests
"""
import pytest
import asyncio
from src.tools.create_task import create_task_internal
from src.tools.update_task import update_task_internal
from tests.utils.task_helpers import create_test_task, count_tasks, get_task_by_id
from tests.utils.agent_helpers import invoke_agent_with_message
@pytest.mark.performance
@pytest.mark.asyncio
async def test_multiple_agents_create_tasks_concurrently_without_conflicts(mock_mcp_context, test_session):
"""
Test: Multiple agents create tasks concurrently without conflicts
Verifies that concurrent task creation doesn't cause database conflicts.
"""
# Define concurrent operations
async def create_task(index):
return await create_task_internal(
ctx=mock_mcp_context,
title=f"Concurrent Task {index}"
)
# Execute 10 concurrent task creations
tasks = [create_task(i) for i in range(10)]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Assert all succeeded
success_count = sum(1 for r in results if not isinstance(r, Exception) and r.get("status") == "success")
assert success_count == 10
# Verify all tasks persisted
task_count = count_tasks(test_session, mock_mcp_context.user_id)
assert task_count == 10
@pytest.mark.performance
@pytest.mark.asyncio
async def test_multiple_agents_update_same_task_concurrently(mock_mcp_context, test_session):
"""
Test: Multiple agents update same task concurrently
Verifies that concurrent updates to same task are handled safely by database.
"""
# Setup: Create a task
task = create_test_task(test_session, mock_mcp_context.user_id, title="Original")
# Define concurrent update operations
async def update_task(suffix):
return await update_task_internal(
ctx=mock_mcp_context,
task_id=task.id,
title=f"Updated {suffix}"
)
# Execute 5 concurrent updates
updates = [update_task(i) for i in range(5)]
results = await asyncio.gather(*updates, return_exceptions=True)
# Assert at least some succeeded (database handles conflicts)
success_count = sum(1 for r in results if not isinstance(r, Exception) and r.get("status") == "success")
assert success_count > 0
# Verify task still exists and has one of the updated titles
updated_task = get_task_by_id(test_session, task.id)
assert updated_task is not None
assert "Updated" in updated_task.title
@pytest.mark.performance
@pytest.mark.slow
@pytest.mark.asyncio
async def test_100_concurrent_chat_requests_complete_successfully(test_user, test_session):
"""
Test: 100 concurrent chat requests complete successfully
Verifies that system handles high concurrent load.
"""
# Define concurrent chat operations
async def send_chat_message(index):
try:
return await invoke_agent_with_message(
user_id=test_user.id,
message=f"Add task number {index}"
)
except Exception as e:
return {"error": str(e)}
# Execute 100 concurrent requests
requests = [send_chat_message(i) for i in range(100)]
results = await asyncio.gather(*requests, return_exceptions=True)
# Assert majority succeeded (allow some failures due to rate limits)
success_count = sum(
1 for r in results
if not isinstance(r, Exception) and "error" not in r
)
# At least 80% should succeed
assert success_count >= 80, f"Only {success_count}/100 requests succeeded"
|