Skip to main content
Glama

MCP Market Statistics Server

by whdghk1907
test_investor_tools.pyβ€’14.4 kB
"""투자자 동ν–₯ 도ꡬ ν…ŒμŠ€νŠΈ""" import pytest from datetime import datetime, timedelta from unittest.mock import AsyncMock from src.tools.investor_tools import InvestorFlowsTool from src.exceptions import DataValidationError, DatabaseConnectionError class TestInvestorFlowsTool: """투자자 동ν–₯ 도ꡬ ν…ŒμŠ€νŠΈ""" @pytest.fixture def mock_db_manager(self): """Mock λ°μ΄ν„°λ² μ΄μŠ€ λ§€λ‹ˆμ €""" return AsyncMock() @pytest.fixture def mock_cache_manager(self): """Mock μΊμ‹œ λ§€λ‹ˆμ €""" return AsyncMock() @pytest.fixture def investor_tool(self, mock_db_manager, mock_cache_manager): """투자자 동ν–₯ 도ꡬ μΈμŠ€ν„΄μŠ€""" return InvestorFlowsTool(mock_db_manager, mock_cache_manager) @pytest.fixture def sample_investor_data(self): """μƒ˜ν”Œ 투자자 데이터""" return [ { "date": datetime.now().date(), "investor_type": "개인", "buy_amount": 1500000000000, # 1.5μ‘° "sell_amount": 1200000000000, # 1.2μ‘° "net_amount": 300000000000, # 3μ²œμ–΅ 순맀수 "market": "KOSPI" }, { "date": datetime.now().date(), "investor_type": "외ꡭ인", "buy_amount": 800000000000, # 8μ²œμ–΅ "sell_amount": 1100000000000, # 1.1μ‘° "net_amount": -300000000000, # 3μ²œμ–΅ μˆœλ§€λ„ "market": "KOSPI" }, { "date": datetime.now().date(), "investor_type": "κΈ°κ΄€", "buy_amount": 700000000000, # 7μ²œμ–΅ "sell_amount": 700000000000, # 7μ²œμ–΅ "net_amount": 0, # κ· ν˜• "market": "KOSPI" } ] def test_tool_initialization(self, investor_tool, mock_db_manager, mock_cache_manager): """도ꡬ μ΄ˆκΈ°ν™” ν…ŒμŠ€νŠΈ""" assert investor_tool.name == "get_investor_flows" assert investor_tool.description is not None assert "투자자" in investor_tool.description assert investor_tool.db_manager == mock_db_manager assert investor_tool.cache_manager == mock_cache_manager def test_tool_definition(self, investor_tool): """도ꡬ μ •μ˜ ν…ŒμŠ€νŠΈ""" definition = investor_tool.get_tool_definition() assert definition.name == "get_investor_flows" 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 "investor_type" in properties # market νŒŒλΌλ―Έν„° 검증 market_prop = properties["market"] assert market_prop["type"] == "string" assert "KOSPI" in market_prop["enum"] assert "KOSDAQ" 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_flows(self, investor_tool, sample_investor_data): """일일 투자자 동ν–₯ 쑰회 ν…ŒμŠ€νŠΈ""" # μΊμ‹œ 미슀 investor_tool.cache_manager.get.return_value = None # λ°μ΄ν„°λ² μ΄μŠ€ 응닡 μ„€μ • investor_tool.db_manager.fetch_all.return_value = sample_investor_data # μ‹€ν–‰ result = await investor_tool.execute({ "market": "KOSPI", "period": "1d", "investor_type": "ALL" }) # κ²°κ³Ό 검증 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 "flows" in data assert "summary" in data # ν”Œλ‘œμš° 데이터 검증 flows = data["flows"] assert len(flows) == 3 # 개인 투자자 데이터 확인 individual = next((f for f in flows if f["investor_type"] == "개인"), None) assert individual is not None assert individual["net_amount"] == 300000000000 assert individual["net_amount_formatted"] == "3,000μ–΅" # 외ꡭ인 데이터 확인 foreign = next((f for f in flows if f["investor_type"] == "외ꡭ인"), None) assert foreign is not None assert foreign["net_amount"] == -300000000000 assert foreign["net_amount_formatted"] == "-3,000μ–΅" # μš”μ•½ 정보 검증 summary = data["summary"] assert "net_individual" in summary assert "net_foreign" in summary assert "net_institution" in summary assert summary["net_individual"] == 300000000000 assert summary["net_foreign"] == -300000000000 assert summary["net_institution"] == 0 @pytest.mark.asyncio async def test_execute_specific_investor_type(self, investor_tool, sample_investor_data): """νŠΉμ • 투자자 μœ ν˜• 쑰회 ν…ŒμŠ€νŠΈ""" # 개인 투자자만 필터링 individual_data = [data for data in sample_investor_data if data["investor_type"] == "개인"] investor_tool.cache_manager.get.return_value = None investor_tool.db_manager.fetch_all.return_value = individual_data # μ‹€ν–‰ result = await investor_tool.execute({ "market": "KOSPI", "period": "1d", "investor_type": "개인" }) # κ²°κ³Ό 검증 content = result[0] import json data = json.loads(content.text) flows = data["flows"] assert len(flows) == 1 assert flows[0]["investor_type"] == "개인" # λ°μ΄ν„°λ² μ΄μŠ€ 쿼리에 ν•„ν„° 쑰건 확인 call_args = investor_tool.db_manager.fetch_all.call_args[0] query = call_args[0] params = call_args[1:] assert "개인" in params @pytest.mark.asyncio async def test_execute_weekly_period(self, investor_tool): """μ£Όκ°„ κΈ°κ°„ 쑰회 ν…ŒμŠ€νŠΈ""" # 일주일 κ°„μ˜ 데이터 weekly_data = [] for i in range(7): date = datetime.now().date() - timedelta(days=i) weekly_data.append({ "date": date, "investor_type": "개인", "buy_amount": 1000000000000, "sell_amount": 900000000000, "net_amount": 100000000000, "market": "KOSPI" }) investor_tool.cache_manager.get.return_value = None investor_tool.db_manager.fetch_all.return_value = weekly_data # μ‹€ν–‰ result = await investor_tool.execute({ "market": "KOSPI", "period": "1w", "investor_type": "개인" }) # κ²°κ³Ό 검증 content = result[0] import json data = json.loads(content.text) assert data["period"] == "1w" # λ°μ΄ν„°λ² μ΄μŠ€ 쿼리에 λ‚ μ§œ λ²”μœ„ 확인 call_args = investor_tool.db_manager.fetch_all.call_args[0] query = call_args[0].upper() assert "DATE >=" in query or ">=" in query assert "7" in str(call_args) or "INTERVAL" in query @pytest.mark.asyncio async def test_cumulative_flows_calculation(self, investor_tool): """λˆ„μ  ν”Œλ‘œμš° 계산 ν…ŒμŠ€νŠΈ""" # 연속 5일 데이터 daily_flows = [] net_amounts = [100, -50, 200, -75, 150] # 얡원 λ‹¨μœ„ for i, net in enumerate(net_amounts): daily_flows.append({ "date": datetime.now().date() - timedelta(days=4-i), "investor_type": "개인", "buy_amount": 1000000000000, "sell_amount": 1000000000000 - (net * 100000000), "net_amount": net * 100000000, # 얡원 -> 원 "market": "KOSPI" }) investor_tool.cache_manager.get.return_value = None investor_tool.db_manager.fetch_all.return_value = daily_flows # μ‹€ν–‰ result = await investor_tool.execute({ "market": "KOSPI", "period": "1w", "investor_type": "개인" }) # κ²°κ³Ό 검증 content = result[0] import json data = json.loads(content.text) assert "cumulative_flows" in data cumulative = data["cumulative_flows"] assert len(cumulative) == 5 # λˆ„μ  계산 확인 (100, 50, 250, 175, 325) expected_cumulative = [100, 50, 250, 175, 325] for i, expected in enumerate(expected_cumulative): assert cumulative[i]["cumulative_net"] == expected * 100000000 @pytest.mark.asyncio async def test_cache_functionality(self, investor_tool, sample_investor_data): """μΊμ‹œ κΈ°λŠ₯ ν…ŒμŠ€νŠΈ""" # μΊμ‹œ 히트 μ‹œλ‚˜λ¦¬μ˜€ cached_data = { "timestamp": datetime.now().isoformat(), "period": "1d", "market": "KOSPI", "flows": [] } investor_tool.cache_manager.get.return_value = cached_data # μ‹€ν–‰ result = await investor_tool.execute({ "market": "KOSPI", "period": "1d" }) # μΊμ‹œμ—μ„œ 데이터 λ°˜ν™˜ 확인 content = result[0] import json data = json.loads(content.text) assert data == cached_data # λ°μ΄ν„°λ² μ΄μŠ€ 호좜 μ—†μŒ 확인 investor_tool.db_manager.fetch_all.assert_not_called() @pytest.mark.asyncio async def test_error_handling(self, investor_tool): """μ—λŸ¬ 처리 ν…ŒμŠ€νŠΈ""" investor_tool.cache_manager.get.return_value = None investor_tool.db_manager.fetch_all.side_effect = DatabaseConnectionError("DB μ—°κ²° μ‹€νŒ¨") with pytest.raises(DatabaseConnectionError): await investor_tool.execute({ "market": "KOSPI", "period": "1d" }) @pytest.mark.asyncio async def test_invalid_parameters(self, investor_tool): """잘λͺ»λœ νŒŒλΌλ―Έν„° ν…ŒμŠ€νŠΈ""" # 잘λͺ»λœ μ‹œμž₯ with pytest.raises(ValueError, match="Invalid market"): await investor_tool.execute({ "market": "INVALID", "period": "1d" }) # 잘λͺ»λœ κΈ°κ°„ with pytest.raises(ValueError, match="Invalid period"): await investor_tool.execute({ "market": "KOSPI", "period": "invalid" }) # 잘λͺ»λœ 투자자 μœ ν˜• with pytest.raises(ValueError, match="Invalid investor_type"): await investor_tool.execute({ "market": "KOSPI", "period": "1d", "investor_type": "invalid" }) @pytest.mark.asyncio async def test_empty_data_handling(self, investor_tool): """빈 데이터 처리 ν…ŒμŠ€νŠΈ""" investor_tool.cache_manager.get.return_value = None investor_tool.db_manager.fetch_all.return_value = [] result = await investor_tool.execute({ "market": "KOSPI", "period": "1d" }) content = result[0] import json data = json.loads(content.text) assert data["flows"] == [] assert "message" in data assert "데이터가 μ—†μŠ΅λ‹ˆλ‹€" in data["message"] or "no data" in data["message"].lower() def test_amount_formatting(self, investor_tool): """κΈˆμ•‘ ν¬λ§·νŒ… ν…ŒμŠ€νŠΈ""" # μ‘° λ‹¨μœ„ formatted = investor_tool._format_amount(1500000000000) assert formatted == "1.5μ‘°" # μ²œμ–΅ λ‹¨μœ„ formatted = investor_tool._format_amount(300000000000) assert formatted == "3,000μ–΅" # λ°±μ–΅ λ‹¨μœ„ formatted = investor_tool._format_amount(50000000000) assert formatted == "500μ–΅" # 음수 formatted = investor_tool._format_amount(-300000000000) assert formatted == "-3,000μ–΅" # 0 formatted = investor_tool._format_amount(0) assert formatted == "0원" @pytest.mark.asyncio async def test_market_impact_analysis(self, investor_tool, sample_investor_data): """μ‹œμž₯ 영ν–₯도 뢄석 ν…ŒμŠ€νŠΈ""" investor_tool.cache_manager.get.return_value = None investor_tool.db_manager.fetch_all.return_value = sample_investor_data # μ‹œμž₯ μ§€μˆ˜ λ³€ν™” 데이터 μΆ”κ°€ Mock investor_tool.db_manager.fetch_one.return_value = { "index_change": 15.30, "index_change_rate": 0.58 } # μ‹€ν–‰ result = await investor_tool.execute({ "market": "KOSPI", "period": "1d", "include_market_impact": True }) # κ²°κ³Ό 검증 content = result[0] import json data = json.loads(content.text) assert "market_impact" in data impact = data["market_impact"] assert "index_change" in impact assert "correlation_analysis" in impact # 상관관계 뢄석 확인 correlation = impact["correlation_analysis"] assert "foreign_vs_index" in correlation # 외ꡭ인 μˆœλ§€λ„(-3μ²œμ–΅)와 μ§€μˆ˜ μƒμŠΉ(+0.58%) κ°„μ˜ 음의 상관관계 assert correlation["foreign_vs_index"] < 0

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