MCP Market Statistics Server

"""시장 개요 도구 테스트""" import pytest from datetime import datetime, timedelta from unittest.mock import Mock, AsyncMock, patch from src.tools.market_tools import MarketOverviewTool from src.exceptions import DataValidationError, DatabaseConnectionError class TestMarketOverviewTool: """시장 개요 도구 테스트""" @pytest.fixture def mock_db_manager(self): """Mock 데이터베이스 매니저""" db_manager = AsyncMock() return db_manager @pytest.fixture def mock_cache_manager(self): """Mock 캐시 매니저""" cache_manager = AsyncMock() return cache_manager @pytest.fixture def market_tool(self, mock_db_manager, mock_cache_manager): """시장 개요 도구 인스턴스""" return MarketOverviewTool(mock_db_manager, mock_cache_manager) @pytest.fixture def sample_market_data(self): """샘플 시장 데이터""" return [ { "index_code": "KOSPI", "current_value": 2650.50, "change": 15.30, "change_rate": 0.58, "volume": 450000000, "market_cap": 1850000000000000, "timestamp": datetime.now() }, { "index_code": "KOSDAQ", "current_value": 850.25, "change": -5.75, "change_rate": -0.67, "volume": 180000000, "market_cap": 285000000000000, "timestamp": datetime.now() } ] def test_tool_initialization(self, market_tool, mock_db_manager, mock_cache_manager): """도구 초기화 테스트""" assert market_tool.name == "get_market_overview" assert market_tool.description is not None assert "시장" in market_tool.description assert market_tool.db_manager == mock_db_manager assert market_tool.cache_manager == mock_cache_manager def test_tool_definition(self, market_tool): """도구 정의 테스트""" definition = market_tool.get_tool_definition() assert definition.name == "get_market_overview" assert definition.description is not None assert definition.inputSchema is not None # 입력 스키마 검증 schema = definition.inputSchema assert schema["type"] == "object" assert "properties" in schema properties = schema["properties"] assert "market" in properties assert "include_details" in properties # market 파라미터 검증 market_prop = properties["market"] assert market_prop["type"] == "string" assert "enum" in market_prop assert "KOSPI" in market_prop["enum"] assert "KOSDAQ" in market_prop["enum"] assert "ALL" in market_prop["enum"] @pytest.mark.asyncio async def test_execute_basic_overview_with_cache_miss(self, market_tool, sample_market_data): """캐시 미스 시 기본 시장 개요 조회 테스트""" # 캐시 미스 설정 market_tool.cache_manager.get.return_value = None # 데이터베이스 응답 설정 market_tool.db_manager.fetch_all.return_value = sample_market_data # 실행 result = await market_tool.execute({"market": "ALL", "include_details": False}) # 결과 검증 assert len(result) == 1 content = result[0] assert content.type == "text" # JSON 파싱하여 내용 확인 import json data = json.loads(content.text) assert "timestamp" in data assert "market_summary" in data assert "indices" in data # 지수 데이터 검증 indices = data["indices"] assert len(indices) == 2 kospi_data = next((idx for idx in indices if idx["index_code"] == "KOSPI"), None) assert kospi_data is not None assert kospi_data["current_value"] == 2650.50 assert kospi_data["change"] == 15.30 assert kospi_data["change_rate"] == 0.58 # 캐시 저장 확인 market_tool.cache_manager.set.assert_called_once() # 데이터베이스 쿼리 확인 market_tool.db_manager.fetch_all.assert_called_once() @pytest.mark.asyncio async def test_execute_with_cache_hit(self, market_tool): """캐시 히트 시 테스트""" # 캐시 데이터 설정 cached_data = { "timestamp": datetime.now().isoformat(), "market_summary": {"total_market_cap": 2000000000000000}, "indices": [{"index_code": "KOSPI", "current_value": 2650.50}] } market_tool.cache_manager.get.return_value = cached_data # 실행 result = await market_tool.execute({"market": "KOSPI"}) # 결과 검증 assert len(result) == 1 content = result[0] import json data = json.loads(content.text) assert data == cached_data # 데이터베이스 호출이 없어야 함 market_tool.db_manager.fetch_all.assert_not_called() # 캐시 저장 호출이 없어야 함 market_tool.cache_manager.set.assert_not_called() @pytest.mark.asyncio async def test_execute_with_details(self, market_tool, sample_market_data): """상세 정보 포함 조회 테스트""" # 캐시 미스 market_tool.cache_manager.get.return_value = None # 메인 데이터 market_tool.db_manager.fetch_all.return_value = sample_market_data # 추가 상세 데이터 Mock market_tool.db_manager.fetch_one.side_effect = [ {"advancing": 850, "declining": 680, "unchanged": 120}, # 시장 폭 데이터 {"new_highs": 45, "new_lows": 23}, # 신고가/신저가 데이터 ] # 실행 result = await market_tool.execute({ "market": "ALL", "include_details": True }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert "market_breadth" in data assert "new_highs_lows" in data # 상세 데이터 검증 breadth = data["market_breadth"] assert breadth["advancing"] == 850 assert breadth["declining"] == 680 highs_lows = data["new_highs_lows"] assert highs_lows["new_highs"] == 45 assert highs_lows["new_lows"] == 23 # 다중 데이터베이스 쿼리 확인 assert market_tool.db_manager.fetch_all.call_count == 1 assert market_tool.db_manager.fetch_one.call_count == 2 @pytest.mark.asyncio async def test_execute_specific_market(self, market_tool, sample_market_data): """특정 시장 조회 테스트""" # KOSPI만 필터링된 데이터 kospi_data = [data for data in sample_market_data if data["index_code"] == "KOSPI"] market_tool.cache_manager.get.return_value = None market_tool.db_manager.fetch_all.return_value = kospi_data # 실행 result = await market_tool.execute({"market": "KOSPI"}) # 결과 검증 content = result[0] import json data = json.loads(content.text) indices = data["indices"] assert len(indices) == 1 assert indices[0]["index_code"] == "KOSPI" # 올바른 WHERE 절로 쿼리되었는지 확인 call_args = market_tool.db_manager.fetch_all.call_args[0] query = call_args[0] assert "WHERE" in query.upper() assert "KOSPI" in call_args[1] or "kospi" in str(call_args).lower() @pytest.mark.asyncio async def test_database_error_handling(self, market_tool): """데이터베이스 오류 처리 테스트""" market_tool.cache_manager.get.return_value = None market_tool.db_manager.fetch_all.side_effect = DatabaseConnectionError("DB 연결 실패") with pytest.raises(DatabaseConnectionError): await market_tool.execute({"market": "ALL"}) @pytest.mark.asyncio async def test_empty_data_handling(self, market_tool): """빈 데이터 처리 테스트""" market_tool.cache_manager.get.return_value = None market_tool.db_manager.fetch_all.return_value = [] result = await market_tool.execute({"market": "ALL"}) content = result[0] import json data = json.loads(content.text) assert data["indices"] == [] assert "message" in data assert "데이터가 없습니다" in data["message"] or "no data" in data["message"].lower() @pytest.mark.asyncio async def test_invalid_market_parameter(self, market_tool): """잘못된 시장 파라미터 테스트""" with pytest.raises(ValueError, match="Invalid market"): await market_tool.execute({"market": "INVALID_MARKET"}) @pytest.mark.asyncio async def test_cache_key_generation(self, market_tool): """캐시 키 생성 테스트""" market_tool.cache_manager.get.return_value = None market_tool.db_manager.fetch_all.return_value = [] await market_tool.execute({"market": "KOSPI", "include_details": True}) # 캐시 조회 시 올바른 키 사용 확인 cache_key = market_tool.cache_manager.get.call_args[0][0] assert "market_overview" in cache_key assert "KOSPI" in cache_key assert "True" in cache_key # include_details @pytest.mark.asyncio async def test_data_freshness_validation(self, market_tool): """데이터 신선도 검증 테스트""" # 5분 전 데이터 (만료되지 않음) fresh_time = datetime.now() - timedelta(minutes=3) market_tool.cache_manager.get.return_value = { "timestamp": fresh_time.isoformat(), "indices": [] } result = await market_tool.execute({"market": "ALL"}) # 캐시 데이터 사용됨 market_tool.db_manager.fetch_all.assert_not_called() # 10분 전 데이터 (만료됨) - 새로운 도구 인스턴스로 테스트 market_tool.cache_manager.reset_mock() market_tool.db_manager.reset_mock() stale_time = datetime.now() - timedelta(minutes=10) market_tool.cache_manager.get.return_value = { "timestamp": stale_time.isoformat(), "indices": [] } market_tool.db_manager.fetch_all.return_value = [] await market_tool.execute({"market": "ALL"}) # 데이터베이스에서 새로 조회됨 market_tool.db_manager.fetch_all.assert_called_once() @pytest.mark.asyncio async def test_concurrent_execution_safety(self, market_tool, sample_market_data): """동시 실행 안전성 테스트""" import asyncio market_tool.cache_manager.get.return_value = None market_tool.db_manager.fetch_all.return_value = sample_market_data # 동시에 여러 요청 실행 tasks = [ market_tool.execute({"market": "ALL"}) for _ in range(5) ] results = await asyncio.gather(*tasks) # 모든 결과가 동일해야 함 assert len(results) == 5 for result in results: assert len(result) == 1 content = result[0] import json data = json.loads(content.text) assert len(data["indices"]) == 2