MCP Market Statistics Server

"""실시간 스트리밍 도구 테스트""" import pytest import asyncio import json import time from datetime import datetime, timedelta from unittest.mock import AsyncMock, MagicMock, patch from typing import Dict, List, Any from src.tools.streaming_tools import RealTimeStreamingTool from src.exceptions import DataValidationError, DatabaseConnectionError class TestRealTimeStreamingTool: """실시간 스트리밍 도구 테스트""" @pytest.fixture def mock_db_manager(self): """Mock 데이터베이스 매니저""" return AsyncMock() @pytest.fixture def mock_cache_manager(self): """Mock 캐시 매니저""" return AsyncMock() @pytest.fixture def mock_kafka_producer(self): """Mock Kafka 프로듀서""" producer = MagicMock() producer.send = AsyncMock() producer.flush = AsyncMock() return producer @pytest.fixture def mock_kafka_consumer(self): """Mock Kafka 컨슈머""" consumer = AsyncMock() return consumer @pytest.fixture def streaming_tool(self, mock_db_manager, mock_cache_manager): """실시간 스트리밍 도구 인스턴스""" return RealTimeStreamingTool(mock_db_manager, mock_cache_manager) @pytest.fixture def sample_market_data(self): """샘플 실시간 시장 데이터""" return { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "index_value": 2650.5, "change": 5.2, "change_rate": 0.196, "volume": 125000000, "transaction_value": 8500000000000, "market_cap": 1850000000000000, "individual_buy": 1200000000, "individual_sell": 1100000000, "institution_buy": 2500000000, "institution_sell": 2400000000, "foreign_buy": 800000000, "foreign_sell": 750000000, "advancing_issues": 485, "declining_issues": 412, "unchanged_issues": 103 } def test_tool_initialization(self, streaming_tool, mock_db_manager, mock_cache_manager): """도구 초기화 테스트""" assert streaming_tool.name == "start_realtime_stream" assert streaming_tool.description is not None assert "실시간" in streaming_tool.description or "real-time" in streaming_tool.description.lower() assert streaming_tool.db_manager == mock_db_manager assert streaming_tool.cache_manager == mock_cache_manager def test_tool_definition(self, streaming_tool): """도구 정의 테스트""" definition = streaming_tool.get_tool_definition() assert definition.name == "start_realtime_stream" 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 "markets" in properties assert "stream_types" in properties assert "update_interval" in properties assert "include_analytics" in properties # stream_types 파라미터 검증 stream_prop = properties["stream_types"] assert stream_prop["type"] == "array" assert "market_data" in str(stream_prop) assert "investor_flow" in str(stream_prop) assert "market_breadth" in str(stream_prop) @pytest.mark.asyncio async def test_execute_market_data_stream(self, streaming_tool, sample_market_data): """시장 데이터 스트림 테스트""" # 캐시 미스 streaming_tool.cache_manager.get.return_value = None # 데이터베이스 응답 설정 streaming_tool.db_manager.fetch_all.return_value = [sample_market_data] # 실행 result = await streaming_tool.execute({ "markets": ["KOSPI"], "stream_types": ["market_data"], "update_interval": 5, "duration": 30 }) # 결과 검증 assert len(result) == 1 content = result[0] assert content.type == "text" # JSON 파싱하여 내용 확인 import json data = json.loads(content.text) assert "stream_info" in data assert "stream_data" in data assert "analytics" in data # 스트림 정보 검증 stream_info = data["stream_info"] assert "stream_id" in stream_info assert "status" in stream_info assert stream_info["status"] == "started" assert "markets" in stream_info assert "KOSPI" in stream_info["markets"] @pytest.mark.asyncio async def test_execute_investor_flow_stream(self, streaming_tool): """투자자 자금 흐름 스트림 테스트""" # 투자자 흐름 데이터 investor_flow_data = [ { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "individual_net": 100000000, "institution_net": -50000000, "foreign_net": 20000000, "pension_net": 10000000, "bank_net": -5000000, "insurance_net": 15000000 } ] streaming_tool.cache_manager.get.return_value = None streaming_tool.db_manager.fetch_all.return_value = investor_flow_data # 실행 result = await streaming_tool.execute({ "markets": ["KOSPI"], "stream_types": ["investor_flow"], "update_interval": 10, "include_analytics": True }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert "stream_data" in data stream_data = data["stream_data"] assert "investor_flow" in stream_data investor_data = stream_data["investor_flow"] assert "individual_net" in investor_data[0] assert "institution_net" in investor_data[0] assert "foreign_net" in investor_data[0] @pytest.mark.asyncio async def test_execute_market_breadth_stream(self, streaming_tool): """시장 폭 스트림 테스트""" breadth_data = [ { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "advancing_issues": 485, "declining_issues": 412, "unchanged_issues": 103, "new_highs": 25, "new_lows": 18, "advance_decline_ratio": 1.177, "advance_volume": 65000000, "decline_volume": 55000000, "total_volume": 125000000 } ] streaming_tool.cache_manager.get.return_value = None streaming_tool.db_manager.fetch_all.return_value = breadth_data # 실행 result = await streaming_tool.execute({ "markets": ["KOSPI", "KOSDAQ"], "stream_types": ["market_breadth"], "update_interval": 15 }) # 결과 검증 content = result[0] import json data = json.loads(content.text) stream_data = data["stream_data"] assert "market_breadth" in stream_data breadth = stream_data["market_breadth"][0] assert "advancing_issues" in breadth assert "declining_issues" in breadth assert "advance_decline_ratio" in breadth @pytest.mark.asyncio async def test_comprehensive_stream(self, streaming_tool, sample_market_data): """종합 스트림 테스트 (모든 데이터 타입)""" streaming_tool.cache_manager.get.return_value = None streaming_tool.db_manager.fetch_all.return_value = [sample_market_data] # 실행 result = await streaming_tool.execute({ "markets": ["KOSPI", "KOSDAQ"], "stream_types": ["market_data", "investor_flow", "market_breadth"], "update_interval": 5, "include_analytics": True, "include_alerts": True, "buffer_size": 100 }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert "stream_info" in data assert "stream_data" in data assert "analytics" in data assert "alerts" in data # 모든 스트림 타입 확인 stream_data = data["stream_data"] assert "market_data" in stream_data assert "investor_flow" in stream_data assert "market_breadth" in stream_data def test_kafka_producer_setup(self, streaming_tool): """Kafka 프로듀서 설정 테스트""" config = { "bootstrap_servers": ["localhost:9092"], "key_serializer": "json", "value_serializer": "json", "acks": "all", "retries": 3 } producer = streaming_tool._setup_kafka_producer(config) assert producer is not None assert hasattr(producer, 'config') assert producer.config["bootstrap_servers"] == ["localhost:9092"] def test_kafka_consumer_setup(self, streaming_tool): """Kafka 컨슈머 설정 테스트""" config = { "bootstrap_servers": ["localhost:9092"], "group_id": "market_analytics", "topics": ["market_data", "investor_flow"], "auto_offset_reset": "latest" } consumer = streaming_tool._setup_kafka_consumer(config) assert consumer is not None assert hasattr(consumer, 'config') assert consumer.config["group_id"] == "market_analytics" assert consumer.config["topics"] == ["market_data", "investor_flow"] @pytest.mark.asyncio async def test_data_buffering_system(self, streaming_tool): """데이터 버퍼링 시스템 테스트""" buffer_config = { "buffer_size": 50, "flush_interval": 10, "compression": "gzip" } buffer = streaming_tool._create_data_buffer(buffer_config) # 테스트 데이터 추가 test_data = { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "value": 2650.5 } await buffer.add(test_data) await buffer.add(test_data) # 버퍼 상태 확인 assert buffer.size() == 2 assert not buffer.is_full() # 버퍼 데이터 조회 buffered_data = await buffer.get_all() assert len(buffered_data) == 2 assert buffered_data[0]["market"] == "KOSPI" @pytest.mark.asyncio async def test_backpressure_handling(self, streaming_tool): """백프레셔 처리 테스트""" # 높은 부하 상황 시뮬레이션 high_load_config = { "max_queue_size": 10, "drop_strategy": "oldest", "backpressure_threshold": 0.8 } queue = streaming_tool._create_backpressure_queue(high_load_config) # 큐에 데이터 추가 (용량 초과) for i in range(15): await queue.put({"id": i, "data": f"test_data_{i}"}) # 백프레셔 상태 확인 assert queue.is_backpressure_active() assert queue.qsize() <= 10 # 최대 크기 초과하지 않음 # 가장 오래된 데이터가 삭제되었는지 확인 oldest_item = await queue.get() assert oldest_item["id"] >= 5 # 처음 5개는 삭제됨 @pytest.mark.asyncio async def test_websocket_connection_manager(self, streaming_tool): """WebSocket 연결 관리 테스트""" connection_manager = streaming_tool._create_websocket_manager() # Mock WebSocket 연결 mock_ws1 = MagicMock() mock_ws1.id = "client_1" mock_ws2 = MagicMock() mock_ws2.id = "client_2" # 클라이언트 연결 추가 await connection_manager.add_connection(mock_ws1) await connection_manager.add_connection(mock_ws2) # 연결 상태 확인 assert connection_manager.get_connection_count() == 2 assert connection_manager.has_connection("client_1") assert connection_manager.has_connection("client_2") # 브로드캐스트 테스트 test_message = {"type": "market_update", "data": {"KOSPI": 2650.5}} sent_count = await connection_manager.broadcast(test_message) assert sent_count == 2 # 연결 제거 await connection_manager.remove_connection("client_1") assert connection_manager.get_connection_count() == 1 assert not connection_manager.has_connection("client_1") @pytest.mark.asyncio async def test_stream_analytics_calculation(self, streaming_tool): """스트림 분석 계산 테스트""" # 시계열 데이터 생성 time_series_data = [] base_time = datetime.now() for i in range(60): # 1분간 5초 간격 데이터 timestamp = base_time + timedelta(seconds=i*5) data_point = { "timestamp": timestamp.isoformat(), "value": 2650 + i * 0.5 + (i % 10) * 2, # 상승 추세 + 노이즈 "volume": 1000000 + i * 50000 } time_series_data.append(data_point) # 실시간 분석 수행 analytics = await streaming_tool._calculate_stream_analytics(time_series_data) assert "trend" in analytics assert "volatility" in analytics assert "momentum" in analytics assert "volume_profile" in analytics # 트렌드 확인 (상승 추세여야 함) assert analytics["trend"]["direction"] in ["up", "bullish"] assert analytics["trend"]["strength"] > 0 # 변동성 확인 assert analytics["volatility"]["current"] >= 0 assert "percentile" in analytics["volatility"] # 모멘텀 확인 assert "rsi" in analytics["momentum"] assert "macd" in analytics["momentum"] @pytest.mark.asyncio async def test_alert_generation(self, streaming_tool): """알림 생성 테스트""" # 알림 트리거 데이터 alert_data = { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "index_value": 2700.0, # 임계값 초과 "change_rate": -3.5, # 급락 "volume": 500000000, # 거래량 급증 "volatility": 0.045 # 높은 변동성 } alerts = await streaming_tool._generate_alerts(alert_data) assert len(alerts) > 0 assert any(alert["type"] == "price_movement" for alert in alerts) assert any(alert["severity"] in ["medium", "high"] for alert in alerts) # 알림 내용 검증 price_alert = next(alert for alert in alerts if alert["type"] == "price_movement") assert "급락" in price_alert["message"] or "decline" in price_alert["message"].lower() assert price_alert["threshold"] is not None assert price_alert["actual_value"] is not None @pytest.mark.asyncio async def test_stream_performance_monitoring(self, streaming_tool): """스트림 성능 모니터링 테스트""" monitor = streaming_tool._create_performance_monitor() # 성능 메트릭 기록 start_time = time.time() for i in range(100): await monitor.record_message_processed(f"msg_{i}", time.time()) if i % 10 == 0: await asyncio.sleep(0.01) # 약간의 지연 추가 end_time = time.time() # 성능 메트릭 확인 metrics = await monitor.get_metrics() assert "messages_per_second" in metrics assert "average_latency" in metrics assert "total_messages" in metrics assert "error_rate" in metrics assert metrics["total_messages"] == 100 assert metrics["messages_per_second"] > 0 assert metrics["error_rate"] == 0 # 에러 없이 처리됨 @pytest.mark.asyncio async def test_stream_recovery_mechanism(self, streaming_tool): """스트림 복구 메커니즘 테스트""" recovery_config = { "max_retries": 3, "retry_delay": 0.1, "circuit_breaker_threshold": 5, "recovery_timeout": 30 } recovery_manager = streaming_tool._create_recovery_manager(recovery_config) # 연결 실패 시뮬레이션 with patch('asyncio.sleep'): # 실제 대기 시간 건너뛰기 # 첫 번째 시도 실패 success = await recovery_manager.attempt_recovery("kafka_connection") assert not success # 첫 시도는 실패 # 재시도 후 성공 (Mock으로 성공 시뮬레이션) with patch.object(recovery_manager, '_test_connection', return_value=True): success = await recovery_manager.attempt_recovery("kafka_connection") assert success # 복구 메트릭 확인 metrics = recovery_manager.get_recovery_metrics() assert "total_failures" in metrics assert "successful_recoveries" in metrics assert metrics["total_failures"] > 0 @pytest.mark.asyncio async def test_data_validation(self, streaming_tool): """데이터 검증 테스트""" # 유효한 데이터 valid_data = { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "index_value": 2650.5, "change_rate": 0.196 } # 유효성 검사 is_valid = streaming_tool._validate_stream_data(valid_data, "market_data") assert is_valid # 무효한 데이터 (필수 필드 누락) invalid_data = { "timestamp": datetime.now().isoformat(), "market": "KOSPI" # index_value 누락 } is_valid = streaming_tool._validate_stream_data(invalid_data, "market_data") assert not is_valid # 잘못된 데이터 타입 wrong_type_data = { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "index_value": "not_a_number", # 문자열이지만 숫자여야 함 "change_rate": 0.196 } is_valid = streaming_tool._validate_stream_data(wrong_type_data, "market_data") assert not is_valid @pytest.mark.asyncio async def test_stream_filtering(self, streaming_tool): """스트림 필터링 테스트""" # 필터 설정 filters = { "markets": ["KOSPI"], "min_volume": 1000000, "max_change_rate": 5.0, "exclude_after_hours": True } # 테스트 데이터 test_data = [ {"market": "KOSPI", "volume": 2000000, "change_rate": 1.5, "trading_session": "regular"}, {"market": "KOSDAQ", "volume": 1500000, "change_rate": 2.0, "trading_session": "regular"}, {"market": "KOSPI", "volume": 500000, "change_rate": 1.0, "trading_session": "regular"}, {"market": "KOSPI", "volume": 2000000, "change_rate": 6.0, "trading_session": "regular"}, {"market": "KOSPI", "volume": 2000000, "change_rate": 1.5, "trading_session": "after_hours"} ] # 필터링 적용 filtered_data = await streaming_tool._apply_stream_filters(test_data, filters) # 결과 검증 (첫 번째 데이터만 모든 조건 만족) assert len(filtered_data) == 1 assert filtered_data[0]["market"] == "KOSPI" assert filtered_data[0]["volume"] >= 1000000 assert filtered_data[0]["change_rate"] <= 5.0 assert filtered_data[0]["trading_session"] == "regular" @pytest.mark.asyncio async def test_cache_functionality(self, streaming_tool): """캐시 기능 테스트""" # 캐시 히트 시나리오 cached_stream_info = { "stream_id": "test_stream_123", "status": "active", "start_time": datetime.now().isoformat(), "markets": ["KOSPI"], "stream_types": ["market_data"] } streaming_tool.cache_manager.get.return_value = cached_stream_info # 실행 result = await streaming_tool.execute({ "markets": ["KOSPI"], "stream_types": ["market_data"], "use_cache": True }) # 캐시에서 데이터 반환 확인 content = result[0] import json data = json.loads(content.text) # 캐시된 데이터가 그대로 반환되는지 확인 assert data.get("stream_id") == "test_stream_123" @pytest.mark.asyncio async def test_error_handling(self, streaming_tool): """에러 처리 테스트""" streaming_tool.cache_manager.get.return_value = None streaming_tool.db_manager.fetch_all.side_effect = DatabaseConnectionError("DB 연결 실패") with pytest.raises(DatabaseConnectionError): await streaming_tool.execute({ "markets": ["KOSPI"], "stream_types": ["market_data"] }) @pytest.mark.asyncio async def test_invalid_parameters(self, streaming_tool): """잘못된 파라미터 테스트""" # 빈 시장 목록 with pytest.raises(ValueError, match="At least one market"): await streaming_tool.execute({ "markets": [], "stream_types": ["market_data"] }) # 빈 스트림 타입 목록 with pytest.raises(ValueError, match="At least one stream type"): await streaming_tool.execute({ "markets": ["KOSPI"], "stream_types": [] }) # 잘못된 업데이트 간격 with pytest.raises(ValueError, match="Invalid update interval"): await streaming_tool.execute({ "markets": ["KOSPI"], "stream_types": ["market_data"], "update_interval": 0 }) def test_stream_id_generation(self, streaming_tool): """스트림 ID 생성 테스트""" markets = ["KOSPI", "KOSDAQ"] stream_types = ["market_data", "investor_flow"] stream_id = streaming_tool._generate_stream_id(markets, stream_types) assert stream_id is not None assert len(stream_id) > 10 # 충분히 긴 ID assert "_" in stream_id or "-" in stream_id # 구분자 포함 # 동일한 파라미터로 다시 생성하면 다른 ID (타임스탬프 때문) stream_id2 = streaming_tool._generate_stream_id(markets, stream_types) assert stream_id != stream_id2 def test_message_serialization(self, streaming_tool): """메시지 직렬화 테스트""" test_message = { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "data": { "index_value": 2650.5, "volume": 125000000, "nested_data": { "field1": "value1", "field2": 123 } } } # JSON 직렬화 serialized = streaming_tool._serialize_message(test_message, "json") assert isinstance(serialized, (str, bytes)) # 역직렬화하여 원본과 비교 deserialized = json.loads(serialized) assert deserialized["market"] == "KOSPI" assert deserialized["data"]["index_value"] == 2650.5 # 압축 직렬화 compressed = streaming_tool._serialize_message(test_message, "json_gzip") assert isinstance(compressed, bytes) assert len(compressed) < len(serialized) # 압축으로 크기 감소