68 lines
2.5 KiB
Python
68 lines
2.5 KiB
Python
|
import pytest
|
||
|
|
from unittest.mock import patch
|
||
|
|
from datetime import datetime
|
||
|
|
from app.services.persistent_session import PersistentSessionManager
|
||
|
|
|
||
|
|
|
||
|
|
class TestPersistentSessionManager:
|
||
|
|
@pytest.fixture(autouse=True)
|
||
|
|
def setup(self, mock_settings, mock_opencode_client):
|
||
|
|
self.manager = PersistentSessionManager()
|
||
|
|
self.manager._sessions = {}
|
||
|
|
self.mock_client = mock_opencode_client
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_session(self):
|
||
|
|
with patch("app.services.persistent_session.opencode_client", self.mock_client):
|
||
|
|
session = await self.manager.create_session("Test Session")
|
||
|
|
|
||
|
|
assert session.id is not None
|
||
|
|
assert session.name == "Test Session"
|
||
|
|
assert session.created_at is not None
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_delete_session(self):
|
||
|
|
with patch("app.services.persistent_session.opencode_client", self.mock_client):
|
||
|
|
session = await self.manager.create_session("Temp Session")
|
||
|
|
session_id = session.id
|
||
|
|
|
||
|
|
result = await self.manager.delete_session(session_id)
|
||
|
|
|
||
|
|
assert result is True
|
||
|
|
assert session_id not in self.manager._sessions
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_message(self):
|
||
|
|
with patch("app.services.persistent_session.opencode_client", self.mock_client):
|
||
|
|
session = await self.manager.create_session("Chat Session")
|
||
|
|
|
||
|
|
result = await self.manager.send_message(session.id, "Hello")
|
||
|
|
|
||
|
|
assert result is not None
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_message_to_nonexistent(self):
|
||
|
|
with patch("app.services.persistent_session.opencode_client", self.mock_client):
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
await self.manager.send_message("nonexistent", "Hello")
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_session(self):
|
||
|
|
with patch("app.services.persistent_session.opencode_client", self.mock_client):
|
||
|
|
session = await self.manager.create_session("Get Test")
|
||
|
|
|
||
|
|
retrieved = self.manager.get_session(session.id)
|
||
|
|
|
||
|
|
assert retrieved is not None
|
||
|
|
assert retrieved.id == session.id
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_sessions(self):
|
||
|
|
with patch("app.services.persistent_session.opencode_client", self.mock_client):
|
||
|
|
await self.manager.create_session("Session 1")
|
||
|
|
await self.manager.create_session("Session 2")
|
||
|
|
|
||
|
|
sessions = self.manager.list_sessions()
|
||
|
|
|
||
|
|
assert len(sessions) == 2
|