tests/test_mcp_servers.py

"""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"])