"""
Integration tests against the real Schwab API.
These tests require valid credentials and a token file.
They are skipped if credentials are not available.
Run with: pytest tests/test_real_api.py -v -s
"""
import os
import pytest
from pathlib import Path
from schwab_mcp.config import settings
from schwab_mcp.auth import TokenManager
from schwab_mcp.client import SchwabClient
from schwab_mcp.tools import account, quotes, options, history, instruments, movers
def get_real_client():
"""Create a real SchwabClient using system credentials."""
try:
cfg = settings()
token_path = Path(cfg.schwab_token_path).expanduser()
if not token_path.exists():
return None
token_manager = TokenManager(
client_id=cfg.schwab_client_id,
client_secret=cfg.schwab_client_secret,
token_path=token_path,
)
# Try to load the token
token = token_manager.load_token()
if not token:
return None
return SchwabClient(token_manager=token_manager, timeout=30)
except Exception as e:
print(f"Failed to create client: {e}")
return None
# Check if we can create a real client
_real_client = None
@pytest.fixture(scope="module")
def real_client():
"""Fixture that provides a real SchwabClient."""
global _real_client
if _real_client is None:
_real_client = get_real_client()
if _real_client is None:
pytest.skip("Real API credentials not available")
return _real_client
# Mark all tests to skip if no credentials
pytestmark = pytest.mark.skipif(
get_real_client() is None,
reason="Real API credentials not available",
)
class TestRealAccountTools:
"""Test account tools against real API."""
@pytest.mark.asyncio
async def test_get_account_real(self, real_client):
"""Test getting real account info."""
result = await account.get_account(real_client, {})
print(f"\n--- Account Info ---")
print(f"Account ID: {result['account_id'][:8]}...")
print(f"Account Type: {result['account_type']}")
print(f"Is Taxable: {result['is_taxable']}")
print(f"Balances: {result['balances']}")
assert "account_id" in result
assert "account_type" in result
assert "balances" in result
assert result["balances"] is not None
@pytest.mark.asyncio
async def test_get_positions_real(self, real_client):
"""Test getting real positions."""
result = await account.get_positions(real_client, {})
print(f"\n--- Positions ---")
print(f"Account: {result['account_id'][:8]}...")
print(f"Number of positions: {len(result['positions'])}")
for pos in result["positions"][:5]: # Show first 5
print(f" {pos['symbol']}: {pos['quantity']} shares, "
f"value=${pos['market_value']:.2f}")
assert "account_id" in result
assert "positions" in result
assert isinstance(result["positions"], list)
class TestRealQuoteTools:
"""Test quote tools against real API."""
@pytest.mark.asyncio
async def test_get_quote_real(self, real_client):
"""Test getting a real quote."""
result = await quotes.get_quote(real_client, {"symbol": "AAPL"})
print(f"\n--- Quote: AAPL ---")
print(f"Last Price: ${result['last_price']}")
print(f"Bid/Ask: ${result['bid']} / ${result['ask']}")
print(f"Day Change: {result['day_change']} ({result['day_change_percent']}%)")
print(f"Volume: {result['volume']:,}")
assert result["symbol"] == "AAPL"
assert result["last_price"] is not None
assert result["last_price"] > 0
@pytest.mark.asyncio
async def test_get_quotes_real(self, real_client):
"""Test getting multiple real quotes."""
symbols = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA"]
result = await quotes.get_quotes(real_client, {"symbols": symbols})
print(f"\n--- Multiple Quotes ---")
for q in result["quotes"]:
if "error" not in q:
print(f" {q['symbol']}: ${q['last_price']:.2f}")
assert "quotes" in result
assert len(result["quotes"]) == len(symbols)
class TestRealOptionTools:
"""Test option tools against real API."""
@pytest.mark.asyncio
async def test_get_option_chain_real(self, real_client):
"""Test getting a real option chain."""
result = await options.get_option_chain(
real_client,
{"symbol": "AAPL", "strike_count": 3},
)
print(f"\n--- Option Chain: AAPL ---")
print(f"Underlying Price: ${result['underlying_price']}")
print(f"Status: {result['status']}")
print(f"Number of Calls: {len(result['calls'])}")
print(f"Number of Puts: {len(result['puts'])}")
if result["calls"]:
call = result["calls"][0]
print(f"\nSample Call: {call['symbol']}")
print(f" Strike: ${call['strike']}, Exp: {call['expiration']}")
print(f" Delta: {call['delta']}, IV: {call['implied_volatility']}")
assert result["symbol"] == "AAPL"
assert result["underlying_price"] is not None
class TestRealHistoryTools:
"""Test price history tools against real API."""
@pytest.mark.asyncio
async def test_get_price_history_real(self, real_client):
"""Test getting real price history."""
result = await history.get_price_history(
real_client,
{
"symbol": "AAPL",
"period_type": "month",
"period": 1,
"frequency_type": "daily",
},
)
print(f"\n--- Price History: AAPL ---")
print(f"Period: {result['period']} {result['period_type']}")
print(f"Frequency: {result['frequency']} {result['frequency_type']}")
print(f"Candle Count: {result['candle_count']}")
if result["candles"]:
first = result["candles"][0]
last = result["candles"][-1]
print(f"\nFirst candle: {first['datetime']}")
print(f" O:{first['open']:.2f} H:{first['high']:.2f} "
f"L:{first['low']:.2f} C:{first['close']:.2f}")
print(f"\nLast candle: {last['datetime']}")
print(f" O:{last['open']:.2f} H:{last['high']:.2f} "
f"L:{last['low']:.2f} C:{last['close']:.2f}")
assert result["symbol"] == "AAPL"
assert result["candle_count"] > 0
assert len(result["candles"]) > 0
class TestRealInstrumentsTools:
"""Test instruments tools against real API."""
@pytest.mark.asyncio
async def test_get_instruments_search_real(self, real_client):
"""Test searching for instruments."""
result = await instruments.get_instruments(
real_client, {"symbol": "AAPL"}
)
print(f"\n--- Instrument Search: AAPL ---")
print(f"Query: {result['query']}")
print(f"Projection: {result['projection']}")
print(f"Results: {result['count']}")
for inst in result["instruments"]:
print(f" {inst['symbol']}: {inst['description']} ({inst['asset_type']})")
assert result["count"] >= 1
@pytest.mark.asyncio
async def test_get_instruments_fundamental_real(self, real_client):
"""Test getting fundamental data."""
result = await instruments.get_instruments(
real_client, {"symbol": "AAPL", "projection": "fundamental"}
)
print(f"\n--- Fundamental Data: AAPL ---")
if result["instruments"]:
inst = result["instruments"][0]
print(f"Symbol: {inst['symbol']}")
print(f"Description: {inst['description']}")
if "fundamental" in inst and inst["fundamental"]:
fund = inst["fundamental"]
print(f"\nFundamentals:")
print(f" P/E Ratio: {fund.get('pe_ratio')}")
print(f" EPS: {fund.get('eps')}")
print(f" Div Yield: {fund.get('div_yield')}%")
print(f" Market Cap: ${fund.get('market_cap'):,.0f}" if fund.get('market_cap') else " Market Cap: N/A")
print(f" 52-Week High: ${fund.get('week_52_high')}")
print(f" 52-Week Low: ${fund.get('week_52_low')}")
print(f" Beta: {fund.get('beta')}")
assert result["count"] >= 1
class TestRealMoversTools:
"""Test movers tools against real API."""
@pytest.mark.asyncio
async def test_get_movers_gainers_real(self, real_client):
"""Test getting top gainers."""
result = await movers.get_movers(
real_client,
{"index": "$DJI", "direction": "up", "change": "percent"},
)
print(f"\n--- Top Gainers: $DJI ---")
print(f"Index: {result['index']}")
print(f"Direction: {result['direction']}, Change: {result['change']}")
print(f"Count: {result['count']}")
for m in result["movers"][:5]:
print(f" {m['symbol']}: ${m['last_price']:.2f} "
f"({m['change_percent']:+.2f}%)")
assert "movers" in result
@pytest.mark.asyncio
async def test_get_movers_losers_real(self, real_client):
"""Test getting top losers."""
result = await movers.get_movers(
real_client,
{"index": "$DJI", "direction": "down", "change": "percent"},
)
print(f"\n--- Top Losers: $DJI ---")
print(f"Count: {result['count']}")
for m in result["movers"][:5]:
print(f" {m['symbol']}: ${m['last_price']:.2f} "
f"({m['change_percent']:+.2f}%)")
assert "movers" in result
@pytest.mark.asyncio
async def test_get_movers_volume_real(self, real_client):
"""Test getting most active by volume."""
result = await movers.get_movers(
real_client,
{"index": "NASDAQ", "direction": "up", "change": "value"},
)
print(f"\n--- Most Active by Volume: NASDAQ ---")
print(f"Count: {result['count']}")
for m in result["movers"][:5]:
vol_str = f"{m['volume']:,}" if m['volume'] else "N/A"
print(f" {m['symbol']}: Vol={vol_str}")
assert "movers" in result
