File size: 4,428 Bytes
246d201 |
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 |
import json
from contextlib import contextmanager
from datetime import datetime
from unittest.mock import MagicMock, patch
import pytest
from openhands.server.routes.manage_conversations import (
delete_conversation,
get_conversation,
search_conversations,
update_conversation,
)
from openhands.storage.data_models.conversation_info import ConversationInfo
from openhands.storage.data_models.conversation_info_result_set import (
ConversationInfoResultSet,
)
from openhands.storage.data_models.conversation_status import ConversationStatus
from openhands.storage.memory import InMemoryFileStore
@contextmanager
def _patch_store():
file_store = InMemoryFileStore()
file_store.write(
'sessions/some_conversation_id/metadata.json',
json.dumps(
{
'title': 'Some Conversation',
'selected_repository': 'foobar',
'conversation_id': 'some_conversation_id',
'github_user_id': '12345',
'created_at': '2025-01-01T00:00:00',
'last_updated_at': '2025-01-01T00:01:00',
}
),
)
with patch(
'openhands.storage.conversation.file_conversation_store.get_file_store',
MagicMock(return_value=file_store),
):
with patch(
'openhands.server.routes.manage_conversations.session_manager.file_store',
file_store,
):
yield
@pytest.mark.asyncio
async def test_search_conversations():
with _patch_store():
result_set = await search_conversations(
MagicMock(state=MagicMock(github_token=''))
)
expected = ConversationInfoResultSet(
results=[
ConversationInfo(
conversation_id='some_conversation_id',
title='Some Conversation',
created_at=datetime.fromisoformat('2025-01-01T00:00:00'),
last_updated_at=datetime.fromisoformat('2025-01-01T00:01:00'),
status=ConversationStatus.STOPPED,
selected_repository='foobar',
)
]
)
assert result_set == expected
@pytest.mark.asyncio
async def test_get_conversation():
with _patch_store():
conversation = await get_conversation(
'some_conversation_id', MagicMock(state=MagicMock(github_token=''))
)
expected = ConversationInfo(
conversation_id='some_conversation_id',
title='Some Conversation',
created_at=datetime.fromisoformat('2025-01-01T00:00:00'),
last_updated_at=datetime.fromisoformat('2025-01-01T00:01:00'),
status=ConversationStatus.STOPPED,
selected_repository='foobar',
)
assert conversation == expected
@pytest.mark.asyncio
async def test_get_missing_conversation():
with _patch_store():
assert (
await get_conversation(
'no_such_conversation', MagicMock(state=MagicMock(github_token=''))
)
is None
)
@pytest.mark.asyncio
async def test_update_conversation():
with _patch_store():
await update_conversation(
MagicMock(state=MagicMock(github_token='')),
'some_conversation_id',
'New Title',
)
conversation = await get_conversation(
'some_conversation_id', MagicMock(state=MagicMock(github_token=''))
)
expected = ConversationInfo(
conversation_id='some_conversation_id',
title='New Title',
created_at=datetime.fromisoformat('2025-01-01T00:00:00'),
last_updated_at=datetime.fromisoformat('2025-01-01T00:01:00'),
status=ConversationStatus.STOPPED,
selected_repository='foobar',
)
assert conversation == expected
@pytest.mark.asyncio
async def test_delete_conversation():
with _patch_store():
await delete_conversation(
'some_conversation_id',
MagicMock(state=MagicMock(github_token='')),
)
conversation = await get_conversation(
'some_conversation_id', MagicMock(state=MagicMock(github_token=''))
)
assert conversation is None
|