MCP Market Statistics Server

"""캐시 관리 테스트""" 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