90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
|
"""Tests for RedisConversationStore.
|
||
|
|
|
||
|
|
Uses fakeredis so no real Redis connection is required.
|
||
|
|
All tests follow the same ConversationStore contract as InMemoryConversationStore.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import fakeredis
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def redis_client():
|
||
|
|
"""Return an in-process fake Redis client."""
|
||
|
|
return fakeredis.FakeRedis()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def store(redis_client):
|
||
|
|
"""Return a RedisConversationStore backed by fake Redis."""
|
||
|
|
from app.infrastructure.session.redis_conversation_store import RedisConversationStore
|
||
|
|
return RedisConversationStore(redis_client=redis_client, timeout_seconds=1800)
|
||
|
|
|
||
|
|
|
||
|
|
def test_create_session_returns_session_with_id(store):
|
||
|
|
"""create_session() must return a ConversationSession with a non-empty session_id."""
|
||
|
|
session = store.create_session()
|
||
|
|
assert session.session_id
|
||
|
|
assert len(session.session_id) > 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_session_returns_same_session(store):
|
||
|
|
"""get_session() must return the previously created session."""
|
||
|
|
session = store.create_session()
|
||
|
|
fetched = store.get_session(session.session_id)
|
||
|
|
assert fetched is not None
|
||
|
|
assert fetched.session_id == session.session_id
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_session_returns_none_for_unknown_id(store):
|
||
|
|
"""get_session() must return None when the session_id does not exist."""
|
||
|
|
assert store.get_session("nonexistent-id") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_save_message_appends_to_session(store):
|
||
|
|
"""save_message() must append a message and return the updated session."""
|
||
|
|
session = store.create_session()
|
||
|
|
updated = store.save_message(session.session_id, role="user", content="Hello")
|
||
|
|
assert updated is not None
|
||
|
|
assert len(updated.messages) == 1
|
||
|
|
assert updated.messages[0].role == "user"
|
||
|
|
assert updated.messages[0].content == "Hello"
|
||
|
|
|
||
|
|
|
||
|
|
def test_save_message_persists_across_lookups(store):
|
||
|
|
"""Messages saved to a session must be visible in subsequent get_session calls."""
|
||
|
|
session = store.create_session()
|
||
|
|
store.save_message(session.session_id, role="user", content="test")
|
||
|
|
fetched = store.get_session(session.session_id)
|
||
|
|
assert fetched is not None
|
||
|
|
assert len(fetched.messages) == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_delete_session_removes_it(store):
|
||
|
|
"""delete_session() must return True and remove the session."""
|
||
|
|
session = store.create_session()
|
||
|
|
assert store.delete_session(session.session_id) is True
|
||
|
|
assert store.get_session(session.session_id) is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_delete_session_returns_false_for_unknown(store):
|
||
|
|
"""delete_session() must return False when the session does not exist."""
|
||
|
|
assert store.delete_session("ghost-id") is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_list_sessions_includes_created_session(store):
|
||
|
|
"""list_sessions() must include all active sessions."""
|
||
|
|
session = store.create_session()
|
||
|
|
ids = [s["session_id"] for s in store.list_sessions()]
|
||
|
|
assert session.session_id in ids
|
||
|
|
|
||
|
|
|
||
|
|
def test_session_expires_after_ttl(redis_client):
|
||
|
|
"""Sessions must disappear after the TTL expires."""
|
||
|
|
from app.infrastructure.session.redis_conversation_store import RedisConversationStore
|
||
|
|
store = RedisConversationStore(redis_client=redis_client, timeout_seconds=1)
|
||
|
|
session = store.create_session()
|
||
|
|
# Simulate TTL expiry by deleting the key directly (fakeredis expire(0) is a no-op).
|
||
|
|
redis_client.delete(f"session:{session.session_id}")
|
||
|
|
assert store.get_session(session.session_id) is None
|