574 lines
20 KiB
Python
574 lines
20 KiB
Python
"""
|
|
Unit tests for Redis Service.
|
|
|
|
These tests use mocking to test the RedisService without requiring
|
|
a real Redis connection.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
import json
|
|
|
|
from redis.exceptions import RedisError, ConnectionError as RedisConnectionError
|
|
|
|
from app.services.redis_service import (
|
|
RedisService,
|
|
RedisServiceError,
|
|
RedisConnectionFailed
|
|
)
|
|
|
|
|
|
class TestRedisServiceInit:
|
|
"""Test RedisService initialization."""
|
|
|
|
@patch('app.services.redis_service.redis.ConnectionPool.from_url')
|
|
@patch('app.services.redis_service.redis.Redis')
|
|
def test_init_success(self, mock_redis_class, mock_pool_from_url):
|
|
"""Test successful initialization."""
|
|
# Setup mocks
|
|
mock_pool = MagicMock()
|
|
mock_pool_from_url.return_value = mock_pool
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = True
|
|
mock_redis_class.return_value = mock_client
|
|
|
|
# Create service
|
|
service = RedisService(redis_url='redis://localhost:6379/0')
|
|
|
|
# Verify
|
|
mock_pool_from_url.assert_called_once()
|
|
mock_redis_class.assert_called_once_with(connection_pool=mock_pool)
|
|
mock_client.ping.assert_called_once()
|
|
assert service.client == mock_client
|
|
|
|
@patch('app.services.redis_service.redis.ConnectionPool.from_url')
|
|
def test_init_connection_failed(self, mock_pool_from_url):
|
|
"""Test initialization fails when Redis is unavailable."""
|
|
mock_pool_from_url.side_effect = RedisConnectionError("Connection refused")
|
|
|
|
with pytest.raises(RedisConnectionFailed) as exc_info:
|
|
RedisService(redis_url='redis://localhost:6379/0')
|
|
|
|
assert "Could not connect to Redis" in str(exc_info.value)
|
|
|
|
def test_init_missing_url(self):
|
|
"""Test initialization fails with missing URL."""
|
|
# Clear environment variable and pass empty string
|
|
with patch.dict('os.environ', {'REDIS_URL': ''}, clear=True):
|
|
with pytest.raises(ValueError) as exc_info:
|
|
RedisService(redis_url='')
|
|
|
|
assert "Redis URL not configured" in str(exc_info.value)
|
|
|
|
|
|
class TestRedisServiceOperations:
|
|
"""Test Redis operations (get, set, delete, exists)."""
|
|
|
|
@pytest.fixture
|
|
def redis_service(self):
|
|
"""Create a RedisService with mocked client."""
|
|
with patch('app.services.redis_service.redis.ConnectionPool.from_url') as mock_pool_from_url:
|
|
with patch('app.services.redis_service.redis.Redis') as mock_redis_class:
|
|
mock_pool = MagicMock()
|
|
mock_pool_from_url.return_value = mock_pool
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = True
|
|
mock_redis_class.return_value = mock_client
|
|
|
|
service = RedisService(redis_url='redis://localhost:6379/0')
|
|
yield service
|
|
|
|
def test_get_existing_key(self, redis_service):
|
|
"""Test getting an existing key."""
|
|
redis_service.client.get.return_value = "test_value"
|
|
|
|
result = redis_service.get("test_key")
|
|
|
|
redis_service.client.get.assert_called_once_with("test_key")
|
|
assert result == "test_value"
|
|
|
|
def test_get_nonexistent_key(self, redis_service):
|
|
"""Test getting a non-existent key returns None."""
|
|
redis_service.client.get.return_value = None
|
|
|
|
result = redis_service.get("nonexistent_key")
|
|
|
|
assert result is None
|
|
|
|
def test_get_error(self, redis_service):
|
|
"""Test get raises RedisServiceError on failure."""
|
|
redis_service.client.get.side_effect = RedisError("Connection lost")
|
|
|
|
with pytest.raises(RedisServiceError) as exc_info:
|
|
redis_service.get("test_key")
|
|
|
|
assert "Failed to get key" in str(exc_info.value)
|
|
|
|
def test_set_basic(self, redis_service):
|
|
"""Test basic set operation."""
|
|
redis_service.client.set.return_value = True
|
|
|
|
result = redis_service.set("test_key", "test_value")
|
|
|
|
redis_service.client.set.assert_called_once_with(
|
|
"test_key", "test_value", ex=None, nx=False, xx=False
|
|
)
|
|
assert result is True
|
|
|
|
def test_set_with_ttl(self, redis_service):
|
|
"""Test set with TTL."""
|
|
redis_service.client.set.return_value = True
|
|
|
|
result = redis_service.set("test_key", "test_value", ttl=3600)
|
|
|
|
redis_service.client.set.assert_called_once_with(
|
|
"test_key", "test_value", ex=3600, nx=False, xx=False
|
|
)
|
|
assert result is True
|
|
|
|
def test_set_nx_success(self, redis_service):
|
|
"""Test set with NX (only if not exists) - success."""
|
|
redis_service.client.set.return_value = True
|
|
|
|
result = redis_service.set("test_key", "test_value", nx=True)
|
|
|
|
redis_service.client.set.assert_called_once_with(
|
|
"test_key", "test_value", ex=None, nx=True, xx=False
|
|
)
|
|
assert result is True
|
|
|
|
def test_set_nx_failure(self, redis_service):
|
|
"""Test set with NX fails when key exists."""
|
|
redis_service.client.set.return_value = None # NX returns None if key exists
|
|
|
|
result = redis_service.set("test_key", "test_value", nx=True)
|
|
|
|
assert result is False
|
|
|
|
def test_set_error(self, redis_service):
|
|
"""Test set raises RedisServiceError on failure."""
|
|
redis_service.client.set.side_effect = RedisError("Connection lost")
|
|
|
|
with pytest.raises(RedisServiceError) as exc_info:
|
|
redis_service.set("test_key", "test_value")
|
|
|
|
assert "Failed to set key" in str(exc_info.value)
|
|
|
|
def test_delete_single_key(self, redis_service):
|
|
"""Test deleting a single key."""
|
|
redis_service.client.delete.return_value = 1
|
|
|
|
result = redis_service.delete("test_key")
|
|
|
|
redis_service.client.delete.assert_called_once_with("test_key")
|
|
assert result == 1
|
|
|
|
def test_delete_multiple_keys(self, redis_service):
|
|
"""Test deleting multiple keys."""
|
|
redis_service.client.delete.return_value = 3
|
|
|
|
result = redis_service.delete("key1", "key2", "key3")
|
|
|
|
redis_service.client.delete.assert_called_once_with("key1", "key2", "key3")
|
|
assert result == 3
|
|
|
|
def test_delete_no_keys(self, redis_service):
|
|
"""Test delete with no keys returns 0."""
|
|
result = redis_service.delete()
|
|
|
|
redis_service.client.delete.assert_not_called()
|
|
assert result == 0
|
|
|
|
def test_delete_error(self, redis_service):
|
|
"""Test delete raises RedisServiceError on failure."""
|
|
redis_service.client.delete.side_effect = RedisError("Connection lost")
|
|
|
|
with pytest.raises(RedisServiceError) as exc_info:
|
|
redis_service.delete("test_key")
|
|
|
|
assert "Failed to delete keys" in str(exc_info.value)
|
|
|
|
def test_exists_single_key(self, redis_service):
|
|
"""Test checking existence of a single key."""
|
|
redis_service.client.exists.return_value = 1
|
|
|
|
result = redis_service.exists("test_key")
|
|
|
|
redis_service.client.exists.assert_called_once_with("test_key")
|
|
assert result == 1
|
|
|
|
def test_exists_multiple_keys(self, redis_service):
|
|
"""Test checking existence of multiple keys."""
|
|
redis_service.client.exists.return_value = 2
|
|
|
|
result = redis_service.exists("key1", "key2", "key3")
|
|
|
|
redis_service.client.exists.assert_called_once_with("key1", "key2", "key3")
|
|
assert result == 2
|
|
|
|
def test_exists_no_keys(self, redis_service):
|
|
"""Test exists with no keys returns 0."""
|
|
result = redis_service.exists()
|
|
|
|
redis_service.client.exists.assert_not_called()
|
|
assert result == 0
|
|
|
|
def test_exists_error(self, redis_service):
|
|
"""Test exists raises RedisServiceError on failure."""
|
|
redis_service.client.exists.side_effect = RedisError("Connection lost")
|
|
|
|
with pytest.raises(RedisServiceError) as exc_info:
|
|
redis_service.exists("test_key")
|
|
|
|
assert "Failed to check existence" in str(exc_info.value)
|
|
|
|
|
|
class TestRedisServiceJSON:
|
|
"""Test JSON serialization methods."""
|
|
|
|
@pytest.fixture
|
|
def redis_service(self):
|
|
"""Create a RedisService with mocked client."""
|
|
with patch('app.services.redis_service.redis.ConnectionPool.from_url') as mock_pool_from_url:
|
|
with patch('app.services.redis_service.redis.Redis') as mock_redis_class:
|
|
mock_pool = MagicMock()
|
|
mock_pool_from_url.return_value = mock_pool
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = True
|
|
mock_redis_class.return_value = mock_client
|
|
|
|
service = RedisService(redis_url='redis://localhost:6379/0')
|
|
yield service
|
|
|
|
def test_get_json_success(self, redis_service):
|
|
"""Test getting and deserializing JSON."""
|
|
test_data = {"name": "test", "value": 123, "nested": {"key": "value"}}
|
|
redis_service.client.get.return_value = json.dumps(test_data)
|
|
|
|
result = redis_service.get_json("test_key")
|
|
|
|
assert result == test_data
|
|
|
|
def test_get_json_none(self, redis_service):
|
|
"""Test get_json returns None for non-existent key."""
|
|
redis_service.client.get.return_value = None
|
|
|
|
result = redis_service.get_json("nonexistent_key")
|
|
|
|
assert result is None
|
|
|
|
def test_get_json_invalid(self, redis_service):
|
|
"""Test get_json raises error for invalid JSON."""
|
|
redis_service.client.get.return_value = "not valid json {"
|
|
|
|
with pytest.raises(RedisServiceError) as exc_info:
|
|
redis_service.get_json("test_key")
|
|
|
|
assert "Failed to decode JSON" in str(exc_info.value)
|
|
|
|
def test_set_json_success(self, redis_service):
|
|
"""Test serializing and setting JSON."""
|
|
redis_service.client.set.return_value = True
|
|
test_data = {"name": "test", "value": 123}
|
|
|
|
result = redis_service.set_json("test_key", test_data, ttl=3600)
|
|
|
|
# Verify the value was serialized
|
|
call_args = redis_service.client.set.call_args
|
|
stored_value = call_args[0][1]
|
|
assert json.loads(stored_value) == test_data
|
|
assert result is True
|
|
|
|
def test_set_json_non_serializable(self, redis_service):
|
|
"""Test set_json raises error for non-serializable data."""
|
|
non_serializable = {"func": lambda x: x}
|
|
|
|
with pytest.raises(RedisServiceError) as exc_info:
|
|
redis_service.set_json("test_key", non_serializable)
|
|
|
|
assert "Failed to serialize value" in str(exc_info.value)
|
|
|
|
|
|
class TestRedisServiceTTL:
|
|
"""Test TTL-related operations."""
|
|
|
|
@pytest.fixture
|
|
def redis_service(self):
|
|
"""Create a RedisService with mocked client."""
|
|
with patch('app.services.redis_service.redis.ConnectionPool.from_url') as mock_pool_from_url:
|
|
with patch('app.services.redis_service.redis.Redis') as mock_redis_class:
|
|
mock_pool = MagicMock()
|
|
mock_pool_from_url.return_value = mock_pool
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = True
|
|
mock_redis_class.return_value = mock_client
|
|
|
|
service = RedisService(redis_url='redis://localhost:6379/0')
|
|
yield service
|
|
|
|
def test_expire_success(self, redis_service):
|
|
"""Test setting expiration on existing key."""
|
|
redis_service.client.expire.return_value = True
|
|
|
|
result = redis_service.expire("test_key", 3600)
|
|
|
|
redis_service.client.expire.assert_called_once_with("test_key", 3600)
|
|
assert result is True
|
|
|
|
def test_expire_nonexistent_key(self, redis_service):
|
|
"""Test expire returns False for non-existent key."""
|
|
redis_service.client.expire.return_value = False
|
|
|
|
result = redis_service.expire("nonexistent_key", 3600)
|
|
|
|
assert result is False
|
|
|
|
def test_ttl_existing_key(self, redis_service):
|
|
"""Test getting TTL of existing key."""
|
|
redis_service.client.ttl.return_value = 3500
|
|
|
|
result = redis_service.ttl("test_key")
|
|
|
|
redis_service.client.ttl.assert_called_once_with("test_key")
|
|
assert result == 3500
|
|
|
|
def test_ttl_no_expiry(self, redis_service):
|
|
"""Test TTL returns -1 for key without expiry."""
|
|
redis_service.client.ttl.return_value = -1
|
|
|
|
result = redis_service.ttl("test_key")
|
|
|
|
assert result == -1
|
|
|
|
def test_ttl_nonexistent_key(self, redis_service):
|
|
"""Test TTL returns -2 for non-existent key."""
|
|
redis_service.client.ttl.return_value = -2
|
|
|
|
result = redis_service.ttl("test_key")
|
|
|
|
assert result == -2
|
|
|
|
|
|
class TestRedisServiceIncrement:
|
|
"""Test increment/decrement operations."""
|
|
|
|
@pytest.fixture
|
|
def redis_service(self):
|
|
"""Create a RedisService with mocked client."""
|
|
with patch('app.services.redis_service.redis.ConnectionPool.from_url') as mock_pool_from_url:
|
|
with patch('app.services.redis_service.redis.Redis') as mock_redis_class:
|
|
mock_pool = MagicMock()
|
|
mock_pool_from_url.return_value = mock_pool
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = True
|
|
mock_redis_class.return_value = mock_client
|
|
|
|
service = RedisService(redis_url='redis://localhost:6379/0')
|
|
yield service
|
|
|
|
def test_incr_default(self, redis_service):
|
|
"""Test incrementing by default amount (1)."""
|
|
redis_service.client.incrby.return_value = 5
|
|
|
|
result = redis_service.incr("counter")
|
|
|
|
redis_service.client.incrby.assert_called_once_with("counter", 1)
|
|
assert result == 5
|
|
|
|
def test_incr_custom_amount(self, redis_service):
|
|
"""Test incrementing by custom amount."""
|
|
redis_service.client.incrby.return_value = 15
|
|
|
|
result = redis_service.incr("counter", 10)
|
|
|
|
redis_service.client.incrby.assert_called_once_with("counter", 10)
|
|
assert result == 15
|
|
|
|
def test_decr_default(self, redis_service):
|
|
"""Test decrementing by default amount (1)."""
|
|
redis_service.client.decrby.return_value = 4
|
|
|
|
result = redis_service.decr("counter")
|
|
|
|
redis_service.client.decrby.assert_called_once_with("counter", 1)
|
|
assert result == 4
|
|
|
|
def test_decr_custom_amount(self, redis_service):
|
|
"""Test decrementing by custom amount."""
|
|
redis_service.client.decrby.return_value = 0
|
|
|
|
result = redis_service.decr("counter", 5)
|
|
|
|
redis_service.client.decrby.assert_called_once_with("counter", 5)
|
|
assert result == 0
|
|
|
|
|
|
class TestRedisServiceHealth:
|
|
"""Test health check and info operations."""
|
|
|
|
@pytest.fixture
|
|
def redis_service(self):
|
|
"""Create a RedisService with mocked client."""
|
|
with patch('app.services.redis_service.redis.ConnectionPool.from_url') as mock_pool_from_url:
|
|
with patch('app.services.redis_service.redis.Redis') as mock_redis_class:
|
|
mock_pool = MagicMock()
|
|
mock_pool_from_url.return_value = mock_pool
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = True
|
|
mock_redis_class.return_value = mock_client
|
|
|
|
service = RedisService(redis_url='redis://localhost:6379/0')
|
|
yield service
|
|
|
|
def test_health_check_success(self, redis_service):
|
|
"""Test health check when Redis is healthy."""
|
|
redis_service.client.ping.return_value = True
|
|
|
|
result = redis_service.health_check()
|
|
|
|
assert result is True
|
|
|
|
def test_health_check_failure(self, redis_service):
|
|
"""Test health check when Redis is unhealthy."""
|
|
redis_service.client.ping.side_effect = RedisError("Connection lost")
|
|
|
|
result = redis_service.health_check()
|
|
|
|
assert result is False
|
|
|
|
def test_info_success(self, redis_service):
|
|
"""Test getting Redis info."""
|
|
mock_info = {
|
|
'redis_version': '7.0.0',
|
|
'used_memory': 1000000,
|
|
'connected_clients': 5
|
|
}
|
|
redis_service.client.info.return_value = mock_info
|
|
|
|
result = redis_service.info()
|
|
|
|
assert result == mock_info
|
|
|
|
def test_info_error(self, redis_service):
|
|
"""Test info raises error on failure."""
|
|
redis_service.client.info.side_effect = RedisError("Connection lost")
|
|
|
|
with pytest.raises(RedisServiceError) as exc_info:
|
|
redis_service.info()
|
|
|
|
assert "Failed to get Redis info" in str(exc_info.value)
|
|
|
|
|
|
class TestRedisServiceUtility:
|
|
"""Test utility methods."""
|
|
|
|
@pytest.fixture
|
|
def redis_service(self):
|
|
"""Create a RedisService with mocked client."""
|
|
with patch('app.services.redis_service.redis.ConnectionPool.from_url') as mock_pool_from_url:
|
|
with patch('app.services.redis_service.redis.Redis') as mock_redis_class:
|
|
mock_pool = MagicMock()
|
|
mock_pool_from_url.return_value = mock_pool
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = True
|
|
mock_redis_class.return_value = mock_client
|
|
|
|
service = RedisService(redis_url='redis://localhost:6379/0')
|
|
yield service
|
|
|
|
def test_flush_db(self, redis_service):
|
|
"""Test flushing database."""
|
|
redis_service.client.flushdb.return_value = True
|
|
|
|
result = redis_service.flush_db()
|
|
|
|
redis_service.client.flushdb.assert_called_once()
|
|
assert result is True
|
|
|
|
def test_close(self, redis_service):
|
|
"""Test closing connection pool."""
|
|
redis_service.close()
|
|
|
|
redis_service.pool.disconnect.assert_called_once()
|
|
|
|
def test_context_manager(self, redis_service):
|
|
"""Test using service as context manager."""
|
|
with redis_service as service:
|
|
assert service is not None
|
|
|
|
redis_service.pool.disconnect.assert_called_once()
|
|
|
|
def test_sanitize_url_with_password(self, redis_service):
|
|
"""Test URL sanitization masks password."""
|
|
url = "redis://user:secretpassword@localhost:6379/0"
|
|
|
|
result = redis_service._sanitize_url(url)
|
|
|
|
assert "secretpassword" not in result
|
|
assert "***" in result
|
|
assert "localhost:6379/0" in result
|
|
|
|
def test_sanitize_url_without_password(self, redis_service):
|
|
"""Test URL sanitization with no password."""
|
|
url = "redis://localhost:6379/0"
|
|
|
|
result = redis_service._sanitize_url(url)
|
|
|
|
assert result == url
|
|
|
|
|
|
class TestRedisServiceIntegration:
|
|
"""Integration-style tests that verify the flow of operations."""
|
|
|
|
@pytest.fixture
|
|
def redis_service(self):
|
|
"""Create a RedisService with mocked client."""
|
|
with patch('app.services.redis_service.redis.ConnectionPool.from_url') as mock_pool_from_url:
|
|
with patch('app.services.redis_service.redis.Redis') as mock_redis_class:
|
|
mock_pool = MagicMock()
|
|
mock_pool_from_url.return_value = mock_pool
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.ping.return_value = True
|
|
mock_redis_class.return_value = mock_client
|
|
|
|
service = RedisService(redis_url='redis://localhost:6379/0')
|
|
yield service
|
|
|
|
def test_set_then_get(self, redis_service):
|
|
"""Test setting and then getting a value."""
|
|
# Set
|
|
redis_service.client.set.return_value = True
|
|
redis_service.set("test_key", "test_value")
|
|
|
|
# Get
|
|
redis_service.client.get.return_value = "test_value"
|
|
result = redis_service.get("test_key")
|
|
|
|
assert result == "test_value"
|
|
|
|
def test_json_roundtrip(self, redis_service):
|
|
"""Test JSON serialization roundtrip."""
|
|
test_data = {
|
|
"user_id": "user_123",
|
|
"tokens_used": 450,
|
|
"model": "claude-3-5-haiku"
|
|
}
|
|
|
|
# Set JSON
|
|
redis_service.client.set.return_value = True
|
|
redis_service.set_json("job_result", test_data, ttl=3600)
|
|
|
|
# Get JSON
|
|
redis_service.client.get.return_value = json.dumps(test_data)
|
|
result = redis_service.get_json("job_result")
|
|
|
|
assert result == test_data
|