MCP Market Statistics Server

"""시장 폭 지표 도구 테스트""" import pytest from datetime import datetime, timedelta from unittest.mock import AsyncMock from src.tools.market_breadth_tools import MarketBreadthTool from src.exceptions import DataValidationError, DatabaseConnectionError class TestMarketBreadthTool: """시장 폭 지표 도구 테스트""" @pytest.fixture def mock_db_manager(self): """Mock 데이터베이스 매니저""" return AsyncMock() @pytest.fixture def mock_cache_manager(self): """Mock 캐시 매니저""" return AsyncMock() @pytest.fixture def breadth_tool(self, mock_db_manager, mock_cache_manager): """시장 폭 지표 도구 인스턴스""" return MarketBreadthTool(mock_db_manager, mock_cache_manager) @pytest.fixture def sample_breadth_data(self): """샘플 시장 폭 데이터""" return [ { "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() }, { "date": datetime.now().date() - timedelta(days=1), "market": "KOSPI", "advancing": 720, "declining": 800, "unchanged": 130, "total_issues": 1650, "advance_decline_ratio": 0.90, "advance_volume": 2100000000000, "decline_volume": 2700000000000, "timestamp": datetime.now() - timedelta(days=1) } ] def test_tool_initialization(self, breadth_tool, mock_db_manager, mock_cache_manager): """도구 초기화 테스트""" assert breadth_tool.name == "get_market_breadth" assert breadth_tool.description is not None assert "시장 폭" in breadth_tool.description assert breadth_tool.db_manager == mock_db_manager assert breadth_tool.cache_manager == mock_cache_manager def test_tool_definition(self, breadth_tool): """도구 정의 테스트""" definition = breadth_tool.get_tool_definition() assert definition.name == "get_market_breadth" 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 "period" in properties assert "include_volume_analysis" in properties # market 파라미터 검증 market_prop = properties["market"] assert market_prop["type"] == "string" assert "KOSPI" in market_prop["enum"] assert "KOSDAQ" in market_prop["enum"] assert "ALL" in market_prop["enum"] # period 파라미터 검증 period_prop = properties["period"] assert period_prop["type"] == "string" assert "1d" in period_prop["enum"] assert "1w" in period_prop["enum"] assert "1m" in period_prop["enum"] @pytest.mark.asyncio async def test_execute_daily_breadth(self, breadth_tool, sample_breadth_data): """일일 시장 폭 조회 테스트""" # 캐시 미스 breadth_tool.cache_manager.get.return_value = None # 데이터베이스 응답 설정 breadth_tool.db_manager.fetch_all.return_value = sample_breadth_data[:1] # 실행 result = await breadth_tool.execute({ "market": "KOSPI", "period": "1d", "include_volume_analysis": 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 "period" in data assert "market" in data assert "breadth_data" in data assert "summary" in data # 시장 폭 데이터 검증 breadth_data = data["breadth_data"] assert len(breadth_data) == 1 today_data = breadth_data[0] assert today_data["advancing"] == 850 assert today_data["declining"] == 680 assert today_data["unchanged"] == 120 assert today_data["advance_decline_ratio"] == 1.25 # 요약 정보 검증 summary = data["summary"] assert "avg_advance_decline_ratio" in summary assert "market_sentiment" in summary assert summary["market_sentiment"] == "상승세" # ratio > 1.0 @pytest.mark.asyncio async def test_execute_weekly_trend(self, breadth_tool, sample_breadth_data): """주간 시장 폭 트렌드 분석 테스트""" # 일주일 간의 데이터 weekly_data = [] for i in range(7): date = datetime.now().date() - timedelta(days=i) advancing = 800 + (i * 20) # 트렌드 생성 declining = 700 - (i * 10) weekly_data.append({ "date": date, "market": "KOSPI", "advancing": advancing, "declining": declining, "unchanged": 150, "total_issues": 1650, "advance_decline_ratio": advancing / declining, "advance_volume": 2500000000000, "decline_volume": 2000000000000, "timestamp": datetime.now() - timedelta(days=i) }) breadth_tool.cache_manager.get.return_value = None breadth_tool.db_manager.fetch_all.return_value = weekly_data # 실행 result = await breadth_tool.execute({ "market": "KOSPI", "period": "1w" }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert data["period"] == "1w" assert len(data["breadth_data"]) == 7 # 트렌드 분석 확인 assert "trend_analysis" in data trend = data["trend_analysis"] assert "direction" in trend assert "strength" in trend @pytest.mark.asyncio async def test_volume_analysis(self, breadth_tool, sample_breadth_data): """거래량 분석 포함 테스트""" breadth_tool.cache_manager.get.return_value = None breadth_tool.db_manager.fetch_all.return_value = sample_breadth_data # 실행 result = await breadth_tool.execute({ "market": "KOSPI", "period": "1d", "include_volume_analysis": True }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert "volume_analysis" in data volume_analysis = data["volume_analysis"] assert "advance_volume" in volume_analysis assert "decline_volume" in volume_analysis assert "volume_ratio" in volume_analysis assert "volume_trend" in volume_analysis @pytest.mark.asyncio async def test_market_sentiment_calculation(self, breadth_tool): """시장 심리 계산 테스트""" # 강한 상승세 데이터 bullish_data = [{ "date": datetime.now().date(), "market": "KOSPI", "advancing": 1200, "declining": 400, "unchanged": 50, "total_issues": 1650, "advance_decline_ratio": 3.0, "advance_volume": 4000000000000, "decline_volume": 1000000000000, "timestamp": datetime.now() }] breadth_tool.cache_manager.get.return_value = None breadth_tool.db_manager.fetch_all.return_value = bullish_data result = await breadth_tool.execute({ "market": "KOSPI", "period": "1d" }) content = result[0] import json data = json.loads(content.text) summary = data["summary"] assert summary["market_sentiment"] == "강한 상승세" # ratio >= 2.0 assert summary["avg_advance_decline_ratio"] == 3.0 def test_advance_decline_ratio_calculation(self, breadth_tool): """상승하락비율 계산 테스트""" # 정상적인 경우 ratio = breadth_tool._calculate_ad_ratio(800, 600) assert abs(ratio - 1.33) < 0.01 # 하락 종목이 0인 경우 ratio = breadth_tool._calculate_ad_ratio(800, 0) assert ratio == float('inf') # 상승 종목이 0인 경우 ratio = breadth_tool._calculate_ad_ratio(0, 600) assert ratio == 0.0 def test_market_sentiment_interpretation(self, breadth_tool): """시장 심리 해석 테스트""" # 강한 상승세 sentiment = breadth_tool._interpret_market_sentiment(2.5) assert sentiment == "강한 상승세" # 상승세 sentiment = breadth_tool._interpret_market_sentiment(1.3) assert sentiment == "상승세" # 보합 sentiment = breadth_tool._interpret_market_sentiment(0.95) assert sentiment == "보합" # 하락세 sentiment = breadth_tool._interpret_market_sentiment(0.6) assert sentiment == "하락세" # 강한 하락세 sentiment = breadth_tool._interpret_market_sentiment(0.3) assert sentiment == "강한 하락세" @pytest.mark.asyncio async def test_cache_functionality(self, breadth_tool): """캐시 기능 테스트""" # 캐시 히트 시나리오 cached_data = { "timestamp": datetime.now().isoformat(), "period": "1d", "market": "KOSPI", "breadth_data": [] } breadth_tool.cache_manager.get.return_value = cached_data # 실행 result = await breadth_tool.execute({ "market": "KOSPI", "period": "1d" }) # 캐시에서 데이터 반환 확인 content = result[0] import json data = json.loads(content.text) assert data == cached_data # 데이터베이스 호출 없음 확인 breadth_tool.db_manager.fetch_all.assert_not_called() @pytest.mark.asyncio async def test_error_handling(self, breadth_tool): """에러 처리 테스트""" breadth_tool.cache_manager.get.return_value = None breadth_tool.db_manager.fetch_all.side_effect = DatabaseConnectionError("DB 연결 실패") with pytest.raises(DatabaseConnectionError): await breadth_tool.execute({ "market": "KOSPI", "period": "1d" }) @pytest.mark.asyncio async def test_invalid_parameters(self, breadth_tool): """잘못된 파라미터 테스트""" # 잘못된 시장 with pytest.raises(ValueError, match="Invalid market"): await breadth_tool.execute({ "market": "INVALID", "period": "1d" }) # 잘못된 기간 with pytest.raises(ValueError, match="Invalid period"): await breadth_tool.execute({ "market": "KOSPI", "period": "invalid" }) @pytest.mark.asyncio async def test_empty_data_handling(self, breadth_tool): """빈 데이터 처리 테스트""" breadth_tool.cache_manager.get.return_value = None breadth_tool.db_manager.fetch_all.return_value = [] result = await breadth_tool.execute({ "market": "KOSPI", "period": "1d" }) content = result[0] import json data = json.loads(content.text) assert data["breadth_data"] == [] assert "message" in data assert "데이터가 없습니다" in data["message"] or "no data" in data["message"].lower() @pytest.mark.asyncio async def test_trend_analysis_calculation(self, breadth_tool): """트렌드 분석 계산 테스트""" # 상승 트렌드 데이터 trend_data = [] for i in range(5): ratio = 0.8 + (i * 0.1) # 0.8 -> 1.2로 상승 트렌드 trend_data.append({ "date": datetime.now().date() - timedelta(days=4-i), "market": "KOSPI", "advancing": int(800 * ratio), "declining": 800, "unchanged": 50, "total_issues": 1650, "advance_decline_ratio": ratio, "advance_volume": 2000000000000, "decline_volume": 2000000000000, "timestamp": datetime.now() - timedelta(days=4-i) }) breadth_tool.cache_manager.get.return_value = None breadth_tool.db_manager.fetch_all.return_value = trend_data result = await breadth_tool.execute({ "market": "KOSPI", "period": "1w" }) content = result[0] import json data = json.loads(content.text) assert "trend_analysis" in data trend = data["trend_analysis"] assert trend["direction"] == "상승" assert "strength" in trend assert trend["days_analyzed"] == 5