"""Unit tests for MCP servers. Tests individual MCP server functionality including: - Yahoo Finance MCP - Financial Modeling Prep MCP - Trading MCP - FRED MCP - Portfolio Optimizer MCP - Risk Analyzer MCP """ import pytest import asyncio from backend.mcp_router import mcp_router class TestYahooFinanceMCP: """Tests for Yahoo Finance MCP server.""" @pytest.mark.asyncio async def test_get_quote(self): """Test fetching real-time quote for single ticker.""" result = await mcp_router.call_yahoo_finance_mcp( "get_quote", {"tickers": ["AAPL"]} ) assert isinstance(result, list) assert len(result) > 0 assert "ticker" in result[0] or "symbol" in result[0] assert "price" in result[0] or "regularMarketPrice" in result[0] @pytest.mark.asyncio async def test_get_quote_multiple_tickers(self): """Test fetching quotes for multiple tickers.""" result = await mcp_router.call_yahoo_finance_mcp( "get_quote", {"tickers": ["AAPL", "MSFT", "GOOGL"]} ) assert isinstance(result, list) assert len(result) == 3 @pytest.mark.asyncio async def test_get_historical_data(self): """Test fetching historical price data.""" result = await mcp_router.call_yahoo_finance_mcp( "get_historical_data", {"ticker": "AAPL", "period": "1mo", "interval": "1d"} ) assert isinstance(result, dict) assert "close_prices" in result assert "dates" in result assert len(result["close_prices"]) > 0 assert len(result["dates"]) > 0 class TestFinancialModelingPrepMCP: """Tests for Financial Modeling Prep MCP server.""" @pytest.mark.asyncio async def test_get_company_profile(self): """Test fetching company profile/fundamentals.""" result = await mcp_router.call_fmp_mcp( "get_company_profile", {"ticker": "AAPL"} ) assert isinstance(result, dict) # Profile should contain at least some company information assert result # Not empty class TestTradingMCP: """Tests for Trading MCP server (technical indicators).""" @pytest.mark.asyncio async def test_get_technical_indicators(self): """Test calculating technical indicators.""" result = await mcp_router.call_trading_mcp( "get_technical_indicators", {"ticker": "AAPL", "period": "3mo"} ) assert isinstance(result, dict) # Should contain at least some indicators expected_keys = {"rsi", "macd", "bollinger_bands", "sma", "ema"} assert any(key in result for key in expected_keys) class TestFREDMCP: """Tests for FRED MCP server (economic data).""" @pytest.mark.asyncio async def test_get_economic_series(self): """Test fetching economic time series.""" result = await mcp_router.call_fred_mcp( "get_economic_series", {"series_id": "GDP"} ) assert isinstance(result, dict) assert "observations" in result or "seriess" in result class TestPortfolioOptimizerMCP: """Tests for Portfolio Optimizer MCP server.""" @pytest.fixture def sample_market_data(self): """Create sample market data for testing.""" return [ { "ticker": "AAPL", "prices": [150.0, 152.0, 151.0, 153.0, 154.0], "dates": ["2024-01-01", "2024-01-02", "2024-01-03", "2024-01-04", "2024-01-05"] }, { "ticker": "MSFT", "prices": [370.0, 372.0, 371.0, 373.0, 374.0], "dates": ["2024-01-01", "2024-01-02", "2024-01-03", "2024-01-04", "2024-01-05"] } ] @pytest.mark.asyncio async def test_optimize_hrp(self, sample_market_data): """Test Hierarchical Risk Parity optimization.""" result = await mcp_router.call_portfolio_optimizer_mcp( "optimize_hrp", { "market_data": sample_market_data, "method": "hrp", "risk_tolerance": "moderate" } ) assert isinstance(result, dict) assert "weights" in result assert isinstance(result["weights"], dict) # Weights should sum to approximately 1.0 weights_sum = sum(result["weights"].values()) assert 0.99 <= weights_sum <= 1.01 @pytest.mark.asyncio async def test_optimize_black_litterman(self, sample_market_data): """Test Black-Litterman optimization.""" result = await mcp_router.call_portfolio_optimizer_mcp( "optimize_black_litterman", { "market_data": sample_market_data, "method": "black_litterman", "risk_tolerance": "moderate" } ) assert isinstance(result, dict) assert "weights" in result @pytest.mark.asyncio async def test_optimize_mean_variance(self, sample_market_data): """Test Mean-Variance (Markowitz) optimization.""" result = await mcp_router.call_portfolio_optimizer_mcp( "optimize_mean_variance", { "market_data": sample_market_data, "method": "mean_variance", "risk_tolerance": "moderate" } ) assert isinstance(result, dict) assert "weights" in result class TestRiskAnalyzerMCP: """Tests for Risk Analyzer MCP server.""" @pytest.fixture def sample_portfolio(self): """Create sample portfolio for testing.""" return [ { "ticker": "AAPL", "weight": 0.6, "prices": [150.0, 152.0, 151.0, 153.0, 154.0, 155.0, 153.0, 156.0] }, { "ticker": "MSFT", "weight": 0.4, "prices": [370.0, 372.0, 371.0, 373.0, 374.0, 375.0, 373.0, 376.0] } ] @pytest.mark.asyncio async def test_analyze_risk(self, sample_portfolio): """Test comprehensive risk analysis.""" result = await mcp_router.call_risk_analyzer_mcp( "analyze_risk", { "portfolio": sample_portfolio, "portfolio_value": 100000.0, "confidence_level": 0.95, "method": "monte_carlo", "num_simulations": 1000 # Reduced for faster tests } ) assert isinstance(result, dict) # Check for VaR and CVaR assert "var_95" in result or "var_99" in result assert "cvar_95" in result or "cvar_99" in result # Check for risk metrics assert "risk_metrics" in result metrics = result["risk_metrics"] assert isinstance(metrics, dict) # Should contain core risk metrics expected_metrics = {"volatility_annual", "sharpe_ratio", "sortino_ratio"} assert any(metric in metrics for metric in expected_metrics) @pytest.mark.asyncio async def test_risk_metrics_validation(self, sample_portfolio): """Test that risk metrics are within reasonable ranges.""" result = await mcp_router.call_risk_analyzer_mcp( "analyze_risk", { "portfolio": sample_portfolio, "portfolio_value": 100000.0, "confidence_level": 0.95, "method": "monte_carlo", "num_simulations": 1000 } ) metrics = result.get("risk_metrics", {}) # Volatility should be positive and reasonable (0-200%) vol = metrics.get("volatility_annual", 0) assert 0 <= vol <= 2.0 # Sharpe ratio should be in reasonable range (-5 to 10) # Note: Values >5 are rare but possible during exceptional market conditions sharpe = metrics.get("sharpe_ratio", 0) assert -5 <= sharpe <= 10 class TestMCPIntegration: """Integration tests for multiple MCP servers working together.""" @pytest.mark.asyncio async def test_complete_workflow(self): """Test complete analysis workflow using multiple MCPs.""" # Step 1: Get market data market_data = await mcp_router.call_yahoo_finance_mcp( "get_quote", {"tickers": ["AAPL", "MSFT"]} ) assert len(market_data) == 2 # Step 2: Get historical data for optimization historical = [] for ticker in ["AAPL", "MSFT"]: hist = await mcp_router.call_yahoo_finance_mcp( "get_historical_data", {"ticker": ticker, "period": "1mo", "interval": "1d"} ) historical.append({ "ticker": ticker, "prices": hist["close_prices"], "dates": hist["dates"] }) # Step 3: Run optimization opt_result = await mcp_router.call_portfolio_optimizer_mcp( "optimize_hrp", { "market_data": historical, "method": "hrp", "risk_tolerance": "moderate" } ) assert "weights" in opt_result # Step 4: Run risk analysis portfolio = [] for i, ticker in enumerate(["AAPL", "MSFT"]): weight = opt_result["weights"].get(ticker, 0.5) portfolio.append({ "ticker": ticker, "weight": weight, "prices": historical[i]["prices"] }) risk_result = await mcp_router.call_risk_analyzer_mcp( "analyze_risk", { "portfolio": portfolio, "portfolio_value": 100000.0, "confidence_level": 0.95, "method": "monte_carlo", "num_simulations": 1000 } ) assert "risk_metrics" in risk_result assert "var_95" in risk_result or "var_99" in risk_result # Pytest configuration @pytest.fixture(scope="session") def event_loop(): """Create event loop for async tests.""" loop = asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() if __name__ == "__main__": pytest.main([__file__, "-v", "-s"])