"""캐시 관리 테스트"""
import pytest
import json
from unittest.mock import Mock, patch, AsyncMock
from src.utils.cache import CacheManager
from src.exceptions import CacheError
class TestCacheManager:
"""캐시 매니저 테스트"""
@pytest.fixture
def cache_config(self):
"""테스트용 캐시 설정"""
return {
"host": "localhost",
"port": 6379,
"db": 0,
"password": None
}
@pytest.fixture
def cache_manager(self, cache_config):
"""테스트용 캐시 매니저"""
return CacheManager(cache_config)
def test_cache_manager_initialization(self, cache_manager, cache_config):
"""캐시 매니저 초기화 테스트"""
assert cache_manager.config == cache_config
assert cache_manager.redis_client is None
assert cache_manager.default_ttl == 300
def test_connection_url_generation(self, cache_manager):
"""Redis 연결 URL 생성 테스트"""
expected_url = "redis://localhost:6379/0"
assert cache_manager.connection_url == expected_url
def test_connection_url_with_password(self):
"""비밀번호가 있는 Redis URL 테스트"""
config = {
"host": "localhost",
"port": 6379,
"db": 1,
"password": "secret123"
}
cache_manager = CacheManager(config)
expected_url = "redis://:secret123@localhost:6379/1"
assert cache_manager.connection_url == expected_url
@pytest.mark.asyncio
async def test_successful_connection(self, cache_manager):
"""성공적인 Redis 연결 테스트"""
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
mock_redis.ping.return_value = True
mock_from_url.return_value = mock_redis
await cache_manager.connect()
assert cache_manager.redis_client == mock_redis
assert cache_manager.is_connected
mock_from_url.assert_called_once()
mock_redis.ping.assert_called_once()
@pytest.mark.asyncio
async def test_connection_failure(self, cache_manager):
"""Redis 연결 실패 테스트"""
with patch('redis.asyncio.from_url') as mock_from_url:
mock_from_url.side_effect = Exception("Connection failed")
with pytest.raises(CacheError):
await cache_manager.connect()
assert cache_manager.redis_client is None
assert not cache_manager.is_connected
@pytest.mark.asyncio
async def test_get_existing_key(self, cache_manager):
"""존재하는 키 조회 테스트"""
test_data = {"message": "hello", "number": 42}
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
mock_redis.ping.return_value = True
mock_redis.get.return_value = json.dumps(test_data)
mock_from_url.return_value = mock_redis
await cache_manager.connect()
result = await cache_manager.get("test_key")
assert result == test_data
mock_redis.get.assert_called_once_with("test_key")
@pytest.mark.asyncio
async def test_get_nonexistent_key(self, cache_manager):
"""존재하지 않는 키 조회 테스트"""
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
mock_redis.ping.return_value = True
mock_redis.get.return_value = None
mock_from_url.return_value = mock_redis
await cache_manager.connect()
result = await cache_manager.get("nonexistent_key")
assert result is None
mock_redis.get.assert_called_once_with("nonexistent_key")
@pytest.mark.asyncio
async def test_set_with_default_ttl(self, cache_manager):
"""기본 TTL로 데이터 저장 테스트"""
test_data = {"message": "hello"}
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
mock_redis.ping.return_value = True
mock_redis.setex.return_value = True
mock_from_url.return_value = mock_redis
await cache_manager.connect()
result = await cache_manager.set("test_key", test_data)
assert result is True
mock_redis.setex.assert_called_once_with(
"test_key",
300, # default TTL
json.dumps(test_data, ensure_ascii=False)
)
@pytest.mark.asyncio
async def test_set_with_custom_ttl(self, cache_manager):
"""커스텀 TTL로 데이터 저장 테스트"""
test_data = {"message": "hello"}
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
mock_redis.ping.return_value = True
mock_redis.setex.return_value = True
mock_from_url.return_value = mock_redis
await cache_manager.connect()
result = await cache_manager.set("test_key", test_data, ttl=600)
assert result is True
mock_redis.setex.assert_called_once_with(
"test_key",
600,
json.dumps(test_data, ensure_ascii=False)
)
@pytest.mark.asyncio
async def test_delete_key(self, cache_manager):
"""키 삭제 테스트"""
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
mock_redis.ping.return_value = True
mock_redis.delete.return_value = 1
mock_from_url.return_value = mock_redis
await cache_manager.connect()
result = await cache_manager.delete("test_key")
assert result is True
mock_redis.delete.assert_called_once_with("test_key")
@pytest.mark.asyncio
async def test_invalidate_pattern(self, cache_manager):
"""패턴 매칭으로 키 무효화 테스트"""
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
mock_redis.ping.return_value = True
mock_redis.keys.return_value = ["market:test1", "market:test2"]
mock_redis.delete.return_value = 2
mock_from_url.return_value = mock_redis
await cache_manager.connect()
await cache_manager.invalidate_pattern("market:*")
mock_redis.keys.assert_called_once_with("market:*")
mock_redis.delete.assert_called_once_with("market:test1", "market:test2")
@pytest.mark.asyncio
async def test_operation_without_connection(self, cache_manager):
"""연결 없이 작업 시 예외 테스트"""
with pytest.raises(CacheError):
await cache_manager.get("test_key")
with pytest.raises(CacheError):
await cache_manager.set("test_key", {"data": "test"})
@pytest.mark.asyncio
async def test_health_check_success(self, cache_manager):
"""헬스체크 성공 테스트"""
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
mock_redis.ping.return_value = True
mock_from_url.return_value = mock_redis
await cache_manager.connect()
is_healthy = await cache_manager.health_check()
assert is_healthy is True
@pytest.mark.asyncio
async def test_health_check_failure(self, cache_manager):
"""헬스체크 실패 테스트"""
with patch('redis.asyncio.from_url') as mock_from_url:
mock_redis = AsyncMock()
# 연결은 성공하지만 헬스체크에서 실패
connect_call = [True] # 첫 호출은 성공
def ping_side_effect(*args, **kwargs):
if connect_call[0]:
connect_call[0] = False
return True # 연결 시 성공
else:
raise Exception("Redis error") # 헬스체크 시 실패
mock_redis.ping.side_effect = ping_side_effect
mock_from_url.return_value = mock_redis
await cache_manager.connect()
is_healthy = await cache_manager.health_check()
assert is_healthy is False
