MCP Market Statistics Server

"""통합 테스트""" import pytest from unittest.mock import AsyncMock from datetime import datetime from src.server import MarketStatsServer from src.tools.market_tools import MarketOverviewTool from src.tools.investor_tools import InvestorFlowsTool from src.tools.market_breadth_tools import MarketBreadthTool from src.tools.sector_tools import SectorAnalysisTool class TestMCPServerIntegration: """MCP 서버 통합 테스트""" @pytest.fixture def mock_dependencies(self): """Mock 의존성들""" mock_db = AsyncMock() mock_cache = AsyncMock() return mock_db, mock_cache @pytest.fixture def server(self, mock_dependencies): """통합 테스트용 서버 인스턴스""" mock_db, mock_cache = mock_dependencies # 서버 설정 config = { "name": "market-stats-server", "version": "1.0.0", "database": {"host": "localhost", "port": 5432, "name": "test_db"}, "redis": {"host": "localhost", "port": 6379} } # 서버 생성 server = MarketStatsServer(config) # Mock 의존성들을 직접 주입 server.db_manager = mock_db server.cache_manager = mock_cache # 모든 도구 등록 market_tool = MarketOverviewTool(mock_db, mock_cache) investor_tool = InvestorFlowsTool(mock_db, mock_cache) breadth_tool = MarketBreadthTool(mock_db, mock_cache) sector_tool = SectorAnalysisTool(mock_db, mock_cache) server.register_tool(market_tool.name, market_tool) server.register_tool(investor_tool.name, investor_tool) server.register_tool(breadth_tool.name, breadth_tool) server.register_tool(sector_tool.name, sector_tool) return server, mock_db, mock_cache def test_all_tools_registered(self, server): """모든 도구가 정상적으로 등록되었는지 테스트""" server_instance, _, _ = server available_tools = server_instance.get_available_tools() tool_names = list(available_tools.keys()) expected_tools = [ "get_market_overview", "get_investor_flows", "get_market_breadth", "get_sector_analysis" ] for expected in expected_tools: assert expected in tool_names assert len(available_tools) == 4 @pytest.mark.asyncio async def test_market_overview_integration(self, server): """시장 개요 도구 통합 테스트""" server_instance, mock_db, mock_cache = server # 서버 시작 await server_instance.startup() try: # Mock 데이터 설정 mock_cache.get.return_value = None mock_db.fetch_all.return_value = [ { "index_code": "KOSPI", "current_value": 2650.50, "change": 15.30, "change_rate": 0.58, "volume": 450000000, "market_cap": 1850000000000000, "timestamp": datetime.now() } ] # 도구 실행 result = await server_instance.execute_tool("get_market_overview", { "market": "KOSPI" }) # 결과 검증 assert len(result) == 1 assert result[0].type == "text" # JSON 검증 import json data = json.loads(result[0].text) assert "timestamp" in data assert "indices" in data assert len(data["indices"]) == 1 assert data["indices"][0]["index_code"] == "KOSPI" finally: await server_instance.shutdown() @pytest.mark.asyncio async def test_investor_flows_integration(self, server): """투자자 동향 도구 통합 테스트""" server_instance, mock_db, mock_cache = server await server_instance.startup() try: # Mock 데이터 설정 mock_cache.get.return_value = None mock_db.fetch_all.return_value = [ { "date": datetime.now().date(), "investor_type": "개인", "buy_amount": 1500000000000, "sell_amount": 1200000000000, "net_amount": 300000000000, "market": "KOSPI" } ] # 도구 실행 result = await server_instance.execute_tool("get_investor_flows", { "market": "KOSPI", "period": "1d" }) # 결과 검증 assert len(result) == 1 import json data = json.loads(result[0].text) assert "flows" in data assert "summary" in data assert len(data["flows"]) == 1 assert data["flows"][0]["investor_type"] == "개인" finally: await server_instance.shutdown() @pytest.mark.asyncio async def test_market_breadth_integration(self, server): """시장 폭 지표 도구 통합 테스트""" server_instance, mock_db, mock_cache = server await server_instance.startup() try: # Mock 데이터 설정 mock_cache.get.return_value = None mock_db.fetch_all.return_value = [ { "date": datetime.now().date(), "market": "KOSPI", "advancing": 850, "declining": 680, "unchanged": 120, "total_issues": 1650, "advance_decline_ratio": 1.25, "advance_volume": 2800000000000, "decline_volume": 1900000000000, "timestamp": datetime.now() } ] # 도구 실행 result = await server_instance.execute_tool("get_market_breadth", { "market": "KOSPI", "period": "1d" }) # 결과 검증 assert len(result) == 1 import json data = json.loads(result[0].text) assert "breadth_data" in data assert "summary" in data assert len(data["breadth_data"]) == 1 assert data["breadth_data"][0]["advancing"] == 850 finally: await server_instance.shutdown() @pytest.mark.asyncio async def test_sector_analysis_integration(self, server): """섹터 분석 도구 통합 테스트""" server_instance, mock_db, mock_cache = server await server_instance.startup() try: # Mock 데이터 설정 mock_cache.get.return_value = None mock_db.fetch_all.return_value = [ { "sector_name": "IT", "sector_code": "IT001", "market_cap": 450000000000000, "total_volume": 280000000000, "change_rate": 2.35, "advancing_stocks": 85, "declining_stocks": 45, "unchanged_stocks": 12, "total_stocks": 142, "top_performer": "삼성전자", "top_performer_change": 3.8, "worst_performer": "LG유플러스", "worst_performer_change": -2.1, "date": datetime.now().date(), "timestamp": datetime.now() } ] # 도구 실행 result = await server_instance.execute_tool("get_sector_analysis", { "market": "ALL", "sector": "ALL" }) # 결과 검증 assert len(result) == 1 import json data = json.loads(result[0].text) assert "sector_analysis" in data assert "market_summary" in data assert len(data["sector_analysis"]) == 1 assert data["sector_analysis"][0]["sector_name"] == "IT" finally: await server_instance.shutdown() @pytest.mark.asyncio async def test_sequential_tool_execution(self, server): """여러 도구 순차 실행 테스트""" server_instance, mock_db, mock_cache = server await server_instance.startup() try: # 모든 도구에 대한 Mock 데이터 설정 mock_cache.get.return_value = None mock_db.fetch_all.side_effect = [ # 시장 개요 데이터 [{"index_code": "KOSPI", "current_value": 2650.50, "change": 15.30, "change_rate": 0.58, "volume": 450000000, "market_cap": 1850000000000000, "timestamp": datetime.now()}], # 투자자 동향 데이터 [{"date": datetime.now().date(), "investor_type": "개인", "buy_amount": 1500000000000, "sell_amount": 1200000000000, "net_amount": 300000000000, "market": "KOSPI"}], # 시장 폭 데이터 [{"date": datetime.now().date(), "market": "KOSPI", "advancing": 850, "declining": 680, "unchanged": 120, "total_issues": 1650, "advance_decline_ratio": 1.25, "advance_volume": 2800000000000, "decline_volume": 1900000000000, "timestamp": datetime.now()}], # 섹터 분석 데이터 [{"sector_name": "IT", "sector_code": "IT001", "market_cap": 450000000000000, "total_volume": 280000000000, "change_rate": 2.35, "advancing_stocks": 85, "declining_stocks": 45, "unchanged_stocks": 12, "total_stocks": 142, "top_performer": "삼성전자", "top_performer_change": 3.8, "worst_performer": "LG유플러스", "worst_performer_change": -2.1, "date": datetime.now().date(), "timestamp": datetime.now()}] ] # 순차적으로 모든 도구 실행 tools = [ ("get_market_overview", {"market": "KOSPI"}), ("get_investor_flows", {"market": "KOSPI", "period": "1d"}), ("get_market_breadth", {"market": "KOSPI", "period": "1d"}), ("get_sector_analysis", {"market": "ALL", "sector": "ALL"}) ] results = [] for tool_name, args in tools: result = await server_instance.execute_tool(tool_name, args) results.append(result) # 모든 결과 검증 assert len(results) == 4 for result in results: assert len(result) == 1 assert result[0].type == "text" # JSON 파싱 가능한지 확인 import json data = json.loads(result[0].text) assert "timestamp" in data finally: await server_instance.shutdown() @pytest.mark.asyncio async def test_error_handling_integration(self, server): """에러 처리 통합 테스트""" server_instance, mock_db, mock_cache = server await server_instance.startup() try: # 존재하지 않는 도구 호출 with pytest.raises(KeyError): await server_instance.execute_tool("nonexistent_tool", {}) # 도구 실행 중 에러 mock_cache.get.return_value = None mock_db.fetch_all.side_effect = Exception("Database error") with pytest.raises(Exception): await server_instance.execute_tool("get_market_overview", {"market": "KOSPI"}) finally: await server_instance.shutdown() @pytest.mark.asyncio async def test_cache_integration(self, server): """캐시 통합 테스트""" server_instance, mock_db, mock_cache = server await server_instance.startup() try: # 캐시 히트 시나리오 cached_data = { "timestamp": datetime.now().isoformat(), "market_summary": {"total_market_cap": 2000000000000000}, "indices": [{"index_code": "KOSPI", "current_value": 2650.50}] } mock_cache.get.return_value = cached_data # 도구 실행 result = await server_instance.execute_tool("get_market_overview", {"market": "KOSPI"}) # 캐시 데이터 반환 확인 import json data = json.loads(result[0].text) assert data == cached_data # 데이터베이스 호출이 없었는지 확인 mock_db.fetch_all.assert_not_called() finally: await server_instance.shutdown() @pytest.mark.asyncio async def test_health_check_integration(self, server): """헬스체크 통합 테스트""" server_instance, mock_db, mock_cache = server await server_instance.startup() try: # 성공 시나리오 mock_db.health_check.return_value = True mock_cache.health_check.return_value = True health_status = await server_instance.health_check() assert health_status["status"] == "healthy" assert health_status["database"] == "connected" assert health_status["cache"] == "connected" # DB 실패 시나리오 mock_db.health_check.return_value = False health_status = await server_instance.health_check() assert health_status["status"] == "unhealthy" finally: await server_instance.shutdown() @pytest.mark.asyncio async def test_server_startup_shutdown_integration(self, server): """서버 시작/종료 통합 테스트""" server_instance, mock_db, mock_cache = server # 초기 상태 확인 assert server_instance.is_running is False # 시작 await server_instance.startup() assert server_instance.is_running is True # 종료 await server_instance.shutdown() assert server_instance.is_running is False # 재시작 테스트 await server_instance.startup() assert server_instance.is_running is True await server_instance.shutdown() assert server_instance.is_running is False