"""Unit tests for unified MCP tools module. Tests the new namespaced tool functions in backend/mcp_tools.py. """ import json import pytest import asyncio from backend import mcp_tools class TestMarketDataTools: """Tests for market data tools (Yahoo Finance).""" @pytest.mark.asyncio async def test_market_get_quote(self): """Test fetching real-time quote for single ticker.""" result = await mcp_tools.market_get_quote(json.dumps(["AAPL"])) assert isinstance(result, list) assert len(result) > 0 assert "ticker" in result[0] or "symbol" in result[0] @pytest.mark.asyncio async def test_market_get_quote_multiple_tickers(self): """Test fetching quotes for multiple tickers.""" result = await mcp_tools.market_get_quote(json.dumps(["AAPL", "MSFT", "GOOGL"])) assert isinstance(result, list) assert len(result) == 3 @pytest.mark.asyncio async def test_market_get_historical_data(self): """Test fetching historical price data.""" result = await mcp_tools.market_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 class TestFundamentalsTools: """Tests for fundamentals tools (FMP).""" @pytest.mark.asyncio async def test_market_get_company_profile(self): """Test fetching company profile.""" result = await mcp_tools.market_get_company_profile(ticker="AAPL") assert isinstance(result, dict) assert result # Not empty class TestTechnicalAnalysisTools: """Tests for technical analysis tools.""" @pytest.mark.asyncio async def test_technical_get_indicators(self): """Test calculating technical indicators.""" result = await mcp_tools.technical_get_indicators(ticker="AAPL", period="3mo") assert isinstance(result, dict) expected_keys = {"rsi", "macd", "bollinger_bands", "moving_averages"} assert any(key in result for key in expected_keys) class TestEconomicDataTools: """Tests for economic data tools (FRED).""" @pytest.mark.asyncio async def test_market_get_economic_series(self): """Test fetching economic time series.""" result = await mcp_tools.market_get_economic_series(series_id="GDP") assert isinstance(result, dict) assert "observations" in result or "series_id" in result class TestPortfolioOptimisationTools: """Tests for portfolio optimisation tools.""" @pytest.fixture def sample_market_data_json(self): """Create sample market data JSON for testing.""" return json.dumps([ { "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_portfolio_optimize_hrp(self, sample_market_data_json): """Test Hierarchical Risk Parity optimisation.""" result = await mcp_tools.portfolio_optimize_hrp( market_data_json=sample_market_data_json, risk_tolerance="moderate" ) assert isinstance(result, dict) assert "weights" in result assert isinstance(result["weights"], dict) weights_sum = sum(float(w) for w in result["weights"].values()) assert 0.99 <= weights_sum <= 1.01 @pytest.mark.asyncio async def test_portfolio_optimize_black_litterman(self, sample_market_data_json): """Test Black-Litterman optimisation.""" result = await mcp_tools.portfolio_optimize_black_litterman( market_data_json=sample_market_data_json, risk_tolerance="moderate" ) assert isinstance(result, dict) assert "weights" in result @pytest.mark.asyncio async def test_portfolio_optimize_mean_variance(self, sample_market_data_json): """Test Mean-Variance (Markowitz) optimisation.""" result = await mcp_tools.portfolio_optimize_mean_variance( market_data_json=sample_market_data_json, risk_tolerance="moderate" ) assert isinstance(result, dict) assert "weights" in result class TestRiskAnalysisTools: """Tests for risk analysis tools.""" @pytest.fixture def sample_portfolio_json(self): """Create sample portfolio JSON for testing.""" return json.dumps([ { "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_risk_analyze(self, sample_portfolio_json): """Test comprehensive risk analysis.""" result = await mcp_tools.risk_analyze( portfolio_json=sample_portfolio_json, portfolio_value="100000", confidence_level="0.95", method="monte_carlo", num_simulations="1000" ) assert isinstance(result, dict) assert "var_95" in result or "var_99" in result assert "cvar_95" in result or "cvar_99" in result assert "risk_metrics" in result class TestCacheBehaviour: """Tests for cache behaviour.""" @pytest.mark.asyncio async def test_cache_hit(self): """Test that caching works correctly.""" result1 = await mcp_tools.market_get_quote(json.dumps(["AAPL"])) result2 = await mcp_tools.market_get_quote(json.dumps(["AAPL"])) assert result1 == result2 class TestConvenienceFunctions: """Tests for internal convenience functions.""" @pytest.mark.asyncio async def test_get_quote_list(self): """Test convenience function with Python list.""" result = await mcp_tools.get_quote_list(["AAPL"]) assert isinstance(result, list) assert len(result) > 0 @pytest.mark.asyncio async def test_get_historical_prices(self): """Test convenience alias function.""" result = await mcp_tools.get_historical_prices("AAPL", "1mo", "1d") assert isinstance(result, dict) assert "close_prices" in result @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"])