Skip to main content
Glama

MCP Market Statistics Server

by whdghk1907
test_collectors.pyβ€’10.1 kB
"""데이터 μˆ˜μ§‘κΈ° ν…ŒμŠ€νŠΈ""" import pytest import asyncio from unittest.mock import Mock, patch, AsyncMock from datetime import datetime, timedelta from src.collectors.base import BaseCollector from src.exceptions import DataCollectionError, DataValidationError class TestBaseCollector: """베이슀 데이터 μˆ˜μ§‘κΈ° ν…ŒμŠ€νŠΈ""" @pytest.fixture def collector_config(self): """ν…ŒμŠ€νŠΈμš© μˆ˜μ§‘κΈ° μ„€μ •""" return { "name": "test_collector", "retry_attempts": 3, "retry_delay": 1, "timeout": 30, "batch_size": 100 } @pytest.fixture def mock_collector(self, collector_config): """ν…ŒμŠ€νŠΈμš© Mock μˆ˜μ§‘κΈ°""" class MockCollector(BaseCollector): async def collect_data(self): """Mock 데이터 μˆ˜μ§‘""" return [ {"symbol": "005930", "price": 75000, "volume": 1000000}, {"symbol": "000660", "price": 45000, "volume": 500000} ] def validate_data(self, data): """Mock 데이터 검증""" for item in data: if not all(k in item for k in ["symbol", "price", "volume"]): return False if item["price"] <= 0 or item["volume"] <= 0: return False return True return MockCollector(collector_config) def test_base_collector_initialization(self, collector_config): """베이슀 μˆ˜μ§‘κΈ° μ΄ˆκΈ°ν™” ν…ŒμŠ€νŠΈ""" class TestCollector(BaseCollector): async def collect_data(self): return [] def validate_data(self, data): return True collector = TestCollector(collector_config) assert collector.name == "test_collector" assert collector.config == collector_config assert collector.retry_attempts == 3 assert collector.retry_delay == 1 assert collector.timeout == 30 assert collector.batch_size == 100 assert collector.is_running is False assert collector.last_run is None assert collector.collected_count == 0 assert collector.error_count == 0 @pytest.mark.asyncio async def test_successful_data_collection(self, mock_collector): """성곡적인 데이터 μˆ˜μ§‘ ν…ŒμŠ€νŠΈ""" result = await mock_collector.run() assert result is not None assert len(result) == 2 assert result[0]["symbol"] == "005930" assert result[1]["symbol"] == "000660" assert mock_collector.collected_count == 2 assert mock_collector.error_count == 0 assert mock_collector.last_run is not None @pytest.mark.asyncio async def test_data_validation_success(self, mock_collector): """데이터 검증 성곡 ν…ŒμŠ€νŠΈ""" test_data = [ {"symbol": "005930", "price": 75000, "volume": 1000000}, {"symbol": "000660", "price": 45000, "volume": 500000} ] is_valid = mock_collector.validate_data(test_data) assert is_valid is True @pytest.mark.asyncio async def test_data_validation_failure(self, mock_collector): """데이터 검증 μ‹€νŒ¨ ν…ŒμŠ€νŠΈ""" invalid_data = [ {"symbol": "005930", "price": -1000, "volume": 1000000}, # 음수 가격 {"symbol": "000660", "price": 45000} # λ³Όλ₯¨ λˆ„λ½ ] is_valid = mock_collector.validate_data(invalid_data) assert is_valid is False @pytest.mark.asyncio async def test_retry_on_failure(self, collector_config): """μ‹€νŒ¨ μ‹œ μž¬μ‹œλ„ ν…ŒμŠ€νŠΈ""" call_count = 0 class FailingCollector(BaseCollector): async def collect_data(self): nonlocal call_count call_count += 1 if call_count <= 2: # 처음 2λ²ˆμ€ μ‹€νŒ¨ raise Exception("Collection failed") return [{"symbol": "005930", "price": 75000, "volume": 1000000}] def validate_data(self, data): return True collector = FailingCollector(collector_config) result = await collector.run() assert call_count == 3 # 2번 μ‹€νŒ¨ ν›„ 3λ²ˆμ§Έμ— 성곡 assert result is not None assert len(result) == 1 assert collector.error_count == 2 # 2번의 μ‹€νŒ¨ assert collector.collected_count == 1 @pytest.mark.asyncio async def test_max_retry_exceeded(self, collector_config): """μ΅œλŒ€ μž¬μ‹œλ„ 초과 ν…ŒμŠ€νŠΈ""" call_count = 0 class AlwaysFailingCollector(BaseCollector): async def collect_data(self): nonlocal call_count call_count += 1 raise Exception("Always fails") def validate_data(self, data): return True collector = AlwaysFailingCollector(collector_config) with pytest.raises(DataCollectionError): await collector.run() assert call_count == 3 # retry_attempts만큼 μ‹œλ„ assert collector.error_count == 3 assert collector.collected_count == 0 @pytest.mark.asyncio async def test_data_validation_error(self, collector_config): """데이터 검증 였λ₯˜ ν…ŒμŠ€νŠΈ""" class InvalidDataCollector(BaseCollector): async def collect_data(self): return [{"symbol": "005930", "price": -1000, "volume": 1000000}] def validate_data(self, data): for item in data: if item["price"] <= 0: return False return True collector = InvalidDataCollector(collector_config) with pytest.raises(DataValidationError): await collector.run() assert collector.collected_count == 0 assert collector.error_count == 1 @pytest.mark.asyncio async def test_batch_processing(self, collector_config): """배치 처리 ν…ŒμŠ€νŠΈ""" collector_config["batch_size"] = 2 # μž‘μ€ 배치 크기둜 ν…ŒμŠ€νŠΈ class BatchCollector(BaseCollector): async def collect_data(self): # 5개 데이터 λ°˜ν™˜ (배치 크기 2보닀 큼) return [ {"symbol": f"00{i:04d}", "price": 1000 + i, "volume": 10000 + i} for i in range(5) ] def validate_data(self, data): return True async def process_batch(self, batch): """배치 처리 μ˜€λ²„λΌμ΄λ“œ""" processed = [] for item in batch: processed.append({ **item, "processed_at": datetime.now().isoformat() }) return processed collector = BatchCollector(collector_config) result = await collector.run() assert len(result) == 5 # 원본 데이터 개수 assert all("processed_at" in item for item in result) # λͺ¨λ“  μ•„μ΄ν…œμ΄ 처리됨 assert collector.collected_count == 5 @pytest.mark.asyncio async def test_concurrent_run_prevention(self, collector_config): """λ™μ‹œ μ‹€ν–‰ λ°©μ§€ ν…ŒμŠ€νŠΈ""" class SlowCollector(BaseCollector): async def collect_data(self): """느린 Mock 데이터 μˆ˜μ§‘""" await asyncio.sleep(0.1) # 0.1초 μ§€μ—° return [{"symbol": "005930", "price": 75000, "volume": 1000000}] def validate_data(self, data): return True collector = SlowCollector(collector_config) # 첫 번째 μ‹€ν–‰ μ‹œμž‘ (μ™„λ£Œλ˜μ§€ μ•ŠμŒ) task1 = asyncio.create_task(collector.run()) await asyncio.sleep(0.05) # task1이 μ‹œμž‘λ˜λ„λ‘ μž μ‹œ λŒ€κΈ° # 두 번째 μ‹€ν–‰ μ‹œλ„ (μ‹€νŒ¨ν•΄μ•Ό 함) with pytest.raises(RuntimeError, match="Collector is already running"): await collector.run() # 첫 번째 μ‹€ν–‰ μ™„λ£Œ result1 = await task1 assert result1 is not None @pytest.mark.asyncio async def test_stats_collection(self, mock_collector): """톡계 μˆ˜μ§‘ ν…ŒμŠ€νŠΈ""" # 첫 번째 μ‹€ν–‰ await mock_collector.run() stats1 = mock_collector.get_stats() assert stats1["total_runs"] == 1 assert stats1["total_collected"] == 2 assert stats1["total_errors"] == 0 assert stats1["success_rate"] == 1.0 # 두 번째 μ‹€ν–‰ (μ‹€νŒ¨ν•˜λ„λ‘ λ§Œλ“€κΈ°) original_collect = mock_collector.collect_data mock_collector.collect_data = AsyncMock(side_effect=Exception("Test error")) try: await mock_collector.run() except DataCollectionError: pass # 톡계 확인 stats2 = mock_collector.get_stats() assert stats2["total_runs"] == 2 assert stats2["total_collected"] == 2 # λ³€ν™” μ—†μŒ assert stats2["total_errors"] == 3 # μž¬μ‹œλ„ 3번으둜 μΈν•œ 3개 μ—λŸ¬ assert stats2["success_rate"] == 0.5 # 1/2 = 50% # μ›λž˜ ν•¨μˆ˜ 볡원 mock_collector.collect_data = original_collect @pytest.mark.asyncio async def test_cleanup_on_shutdown(self, mock_collector): """μ’…λ£Œ μ‹œ 정리 μž‘μ—… ν…ŒμŠ€νŠΈ""" # μ‹€ν–‰ ν›„ μ’…λ£Œ await mock_collector.run() await mock_collector.shutdown() assert mock_collector.is_running is False # 좔가적인 정리 μž‘μ—…μ΄ μžˆλ‹€λ©΄ μ—¬κΈ°μ„œ 확인

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/whdghk1907/mcp-market-statistics'

If you have feedback or need assistance with the MCP directory API, please join our Discord server