"""Unit tests for FutuProvider."""
import pytest
from unittest.mock import MagicMock, patch
import pandas as pd
from datetime import datetime
from stock_data_mcp.providers.futu_provider import FutuProvider
from stock_data_mcp.util import ProviderError
from futu import RET_OK, RET_ERROR, KLType
@pytest.fixture
def mock_context():
"""Mock OpenQuoteContext."""
with patch("stock_data_mcp.providers.futu_provider.OpenQuoteContext") as MockContext:
ctx = MockContext.return_value
yield ctx
@pytest.fixture
def provider(mock_context):
"""Create FutuProvider with mocked context."""
# Prevent caching singleton globally for tests
FutuProvider._instance = None
FutuProvider._context = None
p = FutuProvider()
p._context = mock_context # force inject mock
return p
def test_resolve_symbol(provider):
"""Test symbol resolution."""
# Yahoo -> Futu
assert provider.yahoo_to_futu_code("0700.HK") == "HK.00700"
assert provider.yahoo_to_futu_code("600519.SS") == "SH.600519"
assert provider.yahoo_to_futu_code("000001.SZ") == "SZ.000001"
assert provider.yahoo_to_futu_code("AAPL") == "US.AAPL"
# Futu -> Yahoo
assert provider.futu_to_yahoo_code("HK.00700") == "0700.HK"
assert provider.futu_to_yahoo_code("SH.600519") == "600519.SS"
assert provider.futu_to_yahoo_code("US.AAPL") == "AAPL"
def test_get_quote_success(provider, mock_context):
"""Test successful quote retrieval."""
# Mock return data
mock_df = pd.DataFrame([{
"code": "HK.00700",
"name": "TENCENT",
"last_price": 380.0,
"open_price": 375.0,
"high_price": 385.0,
"low_price": 370.0,
"prev_close_price": 378.0,
"volume": 1000000,
"turnover": 380000000,
"turnover_rate": 0.1,
"amplitude": 2.0,
"suspension": False,
"listing_date": "2004-06-16",
"update_time": "2024-01-01 10:00:00",
"change_val": 2.0,
"change_rate": 0.52
}])
mock_context.get_market_snapshot.return_value = (RET_OK, mock_df)
quote = provider.get_quote("0700.HK")
assert quote.symbol == "0700.HK"
assert quote.price == 380.0
assert quote.market == "HK"
assert quote.source == "futu"
assert quote.source_detail["futu_code"] == "HK.00700"
mock_context.get_market_snapshot.assert_called_with(["HK.00700"])
def test_get_quote_fail(provider, mock_context):
"""Test quote retrieval failure."""
mock_context.get_market_snapshot.return_value = (RET_ERROR, "Mock Error")
with pytest.raises(ProviderError) as exc:
provider.get_quote("0700.HK")
assert exc.value.code == "PROVIDER_ERROR"
assert "Mock Error" in exc.value.message
def test_get_history_pagination(provider, mock_context):
"""Test history with pagination."""
# Mock data pages
df1 = pd.DataFrame([{
"code": "HK.00700",
"time_key": "2024-01-01 00:00:00",
"open": 300, "close": 301, "high": 302, "low": 299, "volume": 1000
}])
df2 = pd.DataFrame([{
"code": "HK.00700",
"time_key": "2024-01-02 00:00:00",
"open": 301, "close": 302, "high": 303, "low": 300, "volume": 1000
}])
# Side effects: (ret, data, page_req_key)
# Call 1: returns df1, key="next"
# Call 2: returns df2, key=None (done)
mock_context.request_history_kline.side_effect = [
(RET_OK, df1, "next_page_key"),
(RET_OK, df2, None)
]
history = provider.get_history("0700.HK", "2024-01-01", "2024-01-02")
assert len(history) == 2
assert history[0].date == "2024-01-01"
assert history[1].date == "2024-01-02"
assert mock_context.request_history_kline.call_count == 2
def test_get_order_book(provider, mock_context):
"""Test order book retrieval."""
mock_data = {
"Bid": [(380.0, 1000, 1)],
"Ask": [(380.2, 2000, 1)]
}
mock_context.get_order_book.return_value = (RET_OK, mock_data)
book = provider.get_order_book("0700.HK")
assert book["symbol"] == "0700.HK"
assert book["futu_code"] == "HK.00700"
assert len(book["bid"]) == 1
assert len(book["ask"]) == 1
def test_subscribe_success(provider, mock_context):
"""Test subscription."""
mock_context.subscribe.return_value = (RET_OK, None)
res = provider.subscribe(["0700.HK"], ["QUOTE"])
assert res["status"] == "success"
# Check if yahoo symbol converted to futu code
args, _ = mock_context.subscribe.call_args
assert args[0] == ["HK.00700"]
