File size: 3,228 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 | """
Integration Tests for Multi-Turn Conversation Flow
Tests conversation context retention and multi-turn interactions.
"""
import pytest
from tests.utils.agent_helpers import invoke_agent_with_message, create_mock_conversation_history
@pytest.mark.integration
@pytest.mark.asyncio
async def test_multi_turn_conversation_with_context_retention(test_user, test_session):
"""
Test: Multi-turn conversation with context retention
Verifies that agent maintains context across multiple turns.
"""
# Turn 1: Create first task
response1 = await invoke_agent_with_message(
user_id=test_user.id,
message="Add a task to call the dentist"
)
# Verify first task created
assert "content" in response1
# Turn 2: Create second task with contextual reference
history = create_mock_conversation_history([
("user", "Add a task to call the dentist"),
("assistant", response1.get("content", "Task created"))
])
response2 = await invoke_agent_with_message(
user_id=test_user.id,
message="Also add one to buy milk",
conversation_history=history
)
# Verify second task created
assert "content" in response2
# Verify both tasks exist in database
from tests.utils.task_helpers import count_tasks
task_count = count_tasks(test_session, test_user.id)
assert task_count >= 2
@pytest.mark.integration
@pytest.mark.asyncio
async def test_agent_references_previous_task_in_conversation(test_user, test_session):
"""
Test: Agent references previous task in conversation
Verifies that agent can reference tasks created earlier in the conversation.
"""
# Turn 1: Create task
response1 = await invoke_agent_with_message(
user_id=test_user.id,
message="Remind me to submit the report"
)
# Turn 2: Reference the task
history = create_mock_conversation_history([
("user", "Remind me to submit the report"),
("assistant", response1.get("content", "Task created"))
])
response2 = await invoke_agent_with_message(
user_id=test_user.id,
message="What tasks do I have?",
conversation_history=history
)
# Verify agent responds with task information
assert "content" in response2
assert len(response2["content"]) > 0
@pytest.mark.integration
@pytest.mark.asyncio
async def test_conversation_maintains_user_context_across_turns(test_user):
"""
Test: Conversation maintains user context across turns
Verifies that user_id context is maintained throughout conversation.
"""
# Turn 1
response1 = await invoke_agent_with_message(
user_id=test_user.id,
message="Create a task to review code"
)
assert "content" in response1
# Turn 2 with history
history = create_mock_conversation_history([
("user", "Create a task to review code"),
("assistant", response1.get("content", "Task created"))
])
response2 = await invoke_agent_with_message(
user_id=test_user.id,
message="Show my tasks",
conversation_history=history
)
# Verify response maintains user context
assert "content" in response2
|