MCP Stock Details Server

""" Test cases for get_analyst_consensus tool (TDD Red Phase) """ import pytest from unittest.mock import AsyncMock, patch, MagicMock from typing import Dict, Any # Mock the modules before importing server import sys from unittest.mock import MagicMock sys.modules['src.tools.analyst_consensus_tools'] = MagicMock() from src.server import MCPStockDetailsServer from src.exceptions import MCPStockDetailsError, InsufficientDataError class TestAnalystConsensus: """Test cases for analyst consensus functionality""" @pytest.fixture def server_with_consensus(self): """Create server instance with analyst consensus capability""" server = MCPStockDetailsServer() server.analyst_consensus_analyzer = AsyncMock() return server @pytest.fixture def sample_consensus_data(self): """Sample analyst consensus data for testing""" return { "005930": { # Samsung Electronics "company_info": { "name": "삼성전자", "current_price": 75000, "currency": "KRW" }, "target_price_consensus": { "mean_target": 82500, "median_target": 83000, "high_target": 95000, "low_target": 70000, "price_upside": 10.0, "analysts_count": 28 }, "investment_opinions": { "buy": 18, "hold": 8, "sell": 2, "total_analysts": 28, "buy_ratio": 64.3, "consensus_rating": "Buy" }, "earnings_estimates": { "current_year": { "revenue_estimate": 305000000000000, "eps_estimate": 4850, "operating_profit_estimate": 48500000000000 }, "next_year": { "revenue_estimate": 315000000000000, "eps_estimate": 5200, "operating_profit_estimate": 52300000000000 } }, "analyst_revisions": { "recent_upgrades": 3, "recent_downgrades": 1, "eps_revisions_up": 12, "eps_revisions_down": 4, "revision_trend": "positive" } } } @pytest.mark.asyncio async def test_get_analyst_consensus_tool_registration(self, server_with_consensus): """Test analyst consensus tool is properly registered""" tools = await server_with_consensus.list_tools() consensus_tool = None for tool in tools: if tool.name == "get_analyst_consensus": consensus_tool = tool break assert consensus_tool is not None assert consensus_tool.description == "애널리스트 컨센서스 및 투자의견" # Check required parameters properties = consensus_tool.inputSchema["properties"] assert "company_code" in properties assert properties["company_code"]["type"] == "string" @pytest.mark.asyncio async def test_analyst_consensus_analysis(self, server_with_consensus, sample_consensus_data): """Test basic analyst consensus analysis""" company_code = "005930" with patch.object(server_with_consensus.analyst_consensus_analyzer, 'get_consensus_data') as mock_data: mock_data.return_value = sample_consensus_data[company_code] result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "include_target_price": True }) assert result is not None content = result[0].text # Should include consensus information assert "ANALYST CONSENSUS" in content assert "삼성전자" in content assert "Target Price" in content assert "82,500" in content or "82500" in content @pytest.mark.asyncio async def test_target_price_consensus(self, server_with_consensus, sample_consensus_data): """Test target price consensus analysis""" company_code = "005930" with patch.object(server_with_consensus.analyst_consensus_analyzer, 'get_consensus_data') as mock_data: mock_data.return_value = sample_consensus_data[company_code] result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "include_target_price": True, "include_price_distribution": True }) assert result is not None content = result[0].text # Should include target price details assert "TARGET PRICE" in content assert "Mean Target" in content or "Average" in content assert "Upside" in content assert "10.0%" in content or "10%" in content @pytest.mark.asyncio async def test_investment_opinions(self, server_with_consensus, sample_consensus_data): """Test investment opinions analysis""" company_code = "005930" with patch.object(server_with_consensus.analyst_consensus_analyzer, 'analyze_investment_opinions') as mock_opinions: mock_opinions.return_value = { "opinion_distribution": { "buy": 18, "hold": 8, "sell": 2, "buy_percentage": 64.3 }, "consensus_strength": "Strong Buy", "opinion_trend": "improving", "key_insights": [ "Strong buy consensus among analysts", "Recent upgrade activity supports positive outlook" ] } result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "include_investment_opinions": True }) assert result is not None content = result[0].text # Should include investment opinions assert "INVESTMENT OPINIONS" in content assert "Buy" in content assert "64.3%" in content or "64%" in content assert "Strong" in content @pytest.mark.asyncio async def test_earnings_estimates(self, server_with_consensus, sample_consensus_data): """Test earnings estimates analysis""" company_code = "005930" with patch.object(server_with_consensus.analyst_consensus_analyzer, 'get_earnings_estimates') as mock_estimates: mock_estimates.return_value = { "current_year_estimates": { "revenue_estimate": 305000000000000, "eps_estimate": 4850, "revenue_growth": 8.5, "eps_growth": 12.3 }, "next_year_estimates": { "revenue_estimate": 315000000000000, "eps_estimate": 5200, "revenue_growth": 3.3, "eps_growth": 7.2 }, "estimate_reliability": "high", "estimate_dispersion": "low" } result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "include_earnings_estimates": True }) assert result is not None content = result[0].text # Should include earnings estimates assert "EARNINGS ESTIMATES" in content assert "Revenue" in content assert "EPS" in content assert "4,850" in content or "4850" in content @pytest.mark.asyncio async def test_analyst_revisions(self, server_with_consensus, sample_consensus_data): """Test analyst revisions tracking""" company_code = "005930" with patch.object(server_with_consensus.analyst_consensus_analyzer, 'track_analyst_revisions') as mock_revisions: mock_revisions.return_value = { "revision_summary": { "recent_upgrades": 3, "recent_downgrades": 1, "net_revisions": "+2", "revision_momentum": "positive" }, "eps_revisions": { "revisions_up": 12, "revisions_down": 4, "net_eps_revisions": "+8", "average_revision": "+2.5%" }, "revision_insights": [ "Positive revision momentum continues", "EPS estimates trending higher" ] } result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "include_revisions": True, "revision_period": "3M" }) assert result is not None content = result[0].text # Should include revisions assert "REVISIONS" in content or "UPGRADES" in content assert "Upgrades" in content or "upgrades" in content assert "positive" in content or "Positive" in content @pytest.mark.asyncio async def test_analyst_coverage_details(self, server_with_consensus, sample_consensus_data): """Test analyst coverage details""" company_code = "005930" with patch.object(server_with_consensus.analyst_consensus_analyzer, 'get_analyst_coverage') as mock_coverage: mock_coverage.return_value = { "coverage_overview": { "total_analysts": 28, "active_coverage": 25, "tier1_analysts": 18, "coverage_quality": "excellent" }, "top_analysts": [ { "analyst_name": "김동원", "firm": "삼성증권", "rating": "Buy", "target_price": 85000, "accuracy_score": 8.5 } ], "coverage_insights": [ "Excellent analyst coverage with 28 analysts", "Strong representation from tier-1 research firms" ] } result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "include_analyst_details": True }) assert result is not None content = result[0].text # Should include coverage details assert "COVERAGE" in content or "ANALYSTS" in content assert "28" in content assert "김동원" in content or "Analyst" in content @pytest.mark.asyncio async def test_earnings_surprise_history(self, server_with_consensus, sample_consensus_data): """Test earnings surprise history""" company_code = "005930" with patch.object(server_with_consensus.analyst_consensus_analyzer, 'analyze_earnings_surprises') as mock_surprises: mock_surprises.return_value = { "surprise_history": [ { "quarter": "2023Q4", "estimated_eps": 4200, "actual_eps": 4500, "surprise_percent": 7.1, "surprise_type": "positive" }, { "quarter": "2023Q3", "estimated_eps": 3800, "actual_eps": 3900, "surprise_percent": 2.6, "surprise_type": "positive" } ], "surprise_statistics": { "average_surprise": 4.9, "positive_surprises": 7, "negative_surprises": 1, "surprise_consistency": "high" } } result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "include_surprise_history": True, "surprise_periods": 8 }) assert result is not None content = result[0].text # Should include surprise history assert "SURPRISE" in content or "EARNINGS" in content assert "2023Q4" in content assert "7.1%" in content or "positive" in content @pytest.mark.asyncio async def test_comprehensive_consensus_report(self, server_with_consensus, sample_consensus_data): """Test comprehensive consensus report""" company_code = "005930" with patch.object(server_with_consensus.analyst_consensus_analyzer, 'comprehensive_consensus_analysis') as mock_comprehensive: mock_comprehensive.return_value = { "consensus_overview": { "overall_sentiment": "Bullish", "confidence_level": "High", "consensus_strength": 8.2 }, "key_themes": [ "Strong semiconductor cycle recovery expected", "Memory market upturn supports outlook", "AI demand driving growth expectations" ], "investment_thesis": [ "Leading position in memory semiconductors", "Beneficiary of AI and data center trends" ], "risk_factors": [ "Cyclical nature of semiconductor business", "Geopolitical tensions affecting supply chain" ] } result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "analysis_type": "comprehensive" }) assert result is not None content = result[0].text # Should include comprehensive analysis assert "COMPREHENSIVE" in content assert "Investment Thesis" in content or "THEMES" in content assert "Bullish" in content or "semiconductor" in content @pytest.mark.asyncio async def test_consensus_changes_tracking(self, server_with_consensus, sample_consensus_data): """Test consensus changes tracking""" company_code = "005930" result = await server_with_consensus._handle_get_analyst_consensus({ "company_code": company_code, "include_consensus_changes": True, "tracking_period": "6M" }) assert result is not None content = result[0].text # Should include consensus changes assert "CONSENSUS" in content or "CHANGES" in content assert "Period" in content or "6M" in content @pytest.mark.asyncio async def test_analyst_consensus_error_handling(self, server_with_consensus): """Test error handling for analyst consensus""" # Test invalid company code with pytest.raises(MCPStockDetailsError): await server_with_consensus._handle_get_analyst_consensus({ "company_code": "", "include_target_price": True }) # Test insufficient consensus data with pytest.raises(InsufficientDataError): await server_with_consensus._handle_get_analyst_consensus({ "company_code": "INVALID", "include_target_price": True })