MCP Market Statistics Server

"""MCP 서버 테스트""" import pytest from unittest.mock import Mock, patch, AsyncMock from src.server import MarketStatsServer class TestMarketStatsServer: """MCP 서버 테스트""" @pytest.fixture def server_config(self): """테스트용 서버 설정""" return { "name": "market-stats-server", "version": "0.1.0", "database": { "host": "localhost", "port": 5432, "name": "test_db", "user": "test_user", "password": "test_pass" }, "redis": { "host": "localhost", "port": 6379, "db": 0, "password": None } } def test_server_initialization(self, server_config): """서버 초기화 테스트""" server = MarketStatsServer(server_config) assert server.name == "market-stats-server" assert server.version == "0.1.0" assert server.config == server_config assert server.tools == {} assert server.db_manager is not None assert server.cache_manager is not None def test_tool_registration(self, server_config): """도구 등록 테스트""" server = MarketStatsServer(server_config) # Mock 도구 생성 mock_tool = Mock() mock_tool.name = "test_tool" mock_tool.description = "Test tool" server.register_tool("test_tool", mock_tool) assert "test_tool" in server.tools assert server.tools["test_tool"] == mock_tool def test_duplicate_tool_registration(self, server_config): """중복 도구 등록 방지 테스트""" server = MarketStatsServer(server_config) mock_tool1 = Mock() mock_tool1.name = "test_tool" mock_tool2 = Mock() mock_tool2.name = "test_tool" server.register_tool("test_tool", mock_tool1) with pytest.raises(ValueError, match="Tool 'test_tool' already registered"): server.register_tool("test_tool", mock_tool2) def test_get_available_tools(self, server_config): """사용 가능한 도구 목록 조회 테스트""" server = MarketStatsServer(server_config) # 여러 도구 등록 tools = ["tool1", "tool2", "tool3"] for tool_name in tools: mock_tool = Mock() mock_tool.name = tool_name server.register_tool(tool_name, mock_tool) available_tools = server.get_available_tools() assert len(available_tools) == 3 assert set(available_tools.keys()) == set(tools) @pytest.mark.asyncio async def test_server_startup(self, server_config): """서버 시작 테스트""" with patch('src.server.DatabaseManager') as mock_db_manager_class: with patch('src.server.CacheManager') as mock_cache_manager_class: mock_db_manager = AsyncMock() mock_cache_manager = AsyncMock() mock_db_manager_class.return_value = mock_db_manager mock_cache_manager_class.return_value = mock_cache_manager server = MarketStatsServer(server_config) await server.startup() mock_db_manager.connect.assert_called_once() mock_cache_manager.connect.assert_called_once() assert server.is_running @pytest.mark.asyncio async def test_server_shutdown(self, server_config): """서버 종료 테스트""" with patch('src.server.DatabaseManager') as mock_db_manager_class: with patch('src.server.CacheManager') as mock_cache_manager_class: mock_db_manager = AsyncMock() mock_cache_manager = AsyncMock() mock_db_manager_class.return_value = mock_db_manager mock_cache_manager_class.return_value = mock_cache_manager server = MarketStatsServer(server_config) await server.startup() await server.shutdown() mock_db_manager.disconnect.assert_called_once() mock_cache_manager.disconnect.assert_called_once() assert not server.is_running @pytest.mark.asyncio async def test_execute_tool_success(self, server_config): """도구 실행 성공 테스트""" server = MarketStatsServer(server_config) # Mock 도구 생성 mock_tool = AsyncMock() mock_tool.execute.return_value = {"result": "success", "data": [1, 2, 3]} server.register_tool("test_tool", mock_tool) result = await server.execute_tool("test_tool", {"param": "value"}) assert result == {"result": "success", "data": [1, 2, 3]} mock_tool.execute.assert_called_once_with({"param": "value"}) @pytest.mark.asyncio async def test_execute_nonexistent_tool(self, server_config): """존재하지 않는 도구 실행 테스트""" server = MarketStatsServer(server_config) with pytest.raises(KeyError, match="Tool 'nonexistent' not found"): await server.execute_tool("nonexistent", {}) @pytest.mark.asyncio async def test_execute_tool_failure(self, server_config): """도구 실행 실패 테스트""" server = MarketStatsServer(server_config) # Mock 도구가 예외 발생 mock_tool = AsyncMock() mock_tool.execute.side_effect = Exception("Tool execution failed") server.register_tool("failing_tool", mock_tool) with pytest.raises(Exception, match="Tool execution failed"): await server.execute_tool("failing_tool", {}) @pytest.mark.asyncio async def test_health_check_success(self, server_config): """헬스체크 성공 테스트""" with patch('src.server.DatabaseManager') as mock_db_manager_class: with patch('src.server.CacheManager') as mock_cache_manager_class: mock_db_manager = AsyncMock() mock_cache_manager = AsyncMock() mock_db_manager.health_check.return_value = True mock_cache_manager.health_check.return_value = True mock_db_manager_class.return_value = mock_db_manager mock_cache_manager_class.return_value = mock_cache_manager server = MarketStatsServer(server_config) await server.startup() health_status = await server.health_check() assert health_status["status"] == "healthy" assert health_status["database"] == "connected" assert health_status["cache"] == "connected" @pytest.mark.asyncio async def test_health_check_database_failure(self, server_config): """데이터베이스 헬스체크 실패 테스트""" with patch('src.server.DatabaseManager') as mock_db_manager_class: with patch('src.server.CacheManager') as mock_cache_manager_class: mock_db_manager = AsyncMock() mock_cache_manager = AsyncMock() mock_db_manager.health_check.return_value = False mock_cache_manager.health_check.return_value = True mock_db_manager_class.return_value = mock_db_manager mock_cache_manager_class.return_value = mock_cache_manager server = MarketStatsServer(server_config) await server.startup() health_status = await server.health_check() assert health_status["status"] == "unhealthy" assert health_status["database"] == "disconnected" assert health_status["cache"] == "connected"