ec005c91a9
All checks were successful
continuous-integration/drone/push Build is passing
- Remove unused datetime imports from openbb_service, market_service, quantitative_service (now using obb_utils.days_ago) - Remove unused variable 'maintains' in routes_sentiment - Remove unused imports in test files - Fix forward reference annotation in test helper
443 lines
16 KiB
Python
443 lines
16 KiB
Python
"""Unit tests for akshare_service.py - written FIRST (TDD RED phase)."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
import akshare_service
|
|
|
|
|
|
# --- Helpers ---
|
|
|
|
|
|
def _make_hist_df(rows: int = 3) -> pd.DataFrame:
|
|
"""Return a minimal historical DataFrame with Chinese column names."""
|
|
dates = pd.date_range("2026-01-01", periods=rows, freq="D")
|
|
return pd.DataFrame(
|
|
{
|
|
"日期": dates,
|
|
"开盘": [10.0] * rows,
|
|
"收盘": [10.5] * rows,
|
|
"最高": [11.0] * rows,
|
|
"最低": [9.5] * rows,
|
|
"成交量": [1_000_000] * rows,
|
|
"成交额": [10_500_000.0] * rows,
|
|
"振幅": [1.5] * rows,
|
|
"涨跌幅": [0.5] * rows,
|
|
"涨跌额": [0.05] * rows,
|
|
"换手率": [0.3] * rows,
|
|
}
|
|
)
|
|
|
|
|
|
def _make_spot_df(code: str = "000001", name: str = "平安银行") -> pd.DataFrame:
|
|
"""Return a minimal real-time quote DataFrame with Chinese column names."""
|
|
return pd.DataFrame(
|
|
[
|
|
{
|
|
"代码": code,
|
|
"名称": name,
|
|
"最新价": 12.34,
|
|
"涨跌幅": 1.23,
|
|
"涨跌额": 0.15,
|
|
"成交量": 500_000,
|
|
"成交额": 6_170_000.0,
|
|
"今开": 12.10,
|
|
"最高": 12.50,
|
|
"最低": 12.00,
|
|
"昨收": 12.19,
|
|
}
|
|
]
|
|
)
|
|
|
|
|
|
def _make_code_name_df() -> pd.DataFrame:
|
|
"""Return a minimal code/name mapping DataFrame."""
|
|
return pd.DataFrame(
|
|
[
|
|
{"code": "000001", "name": "平安银行"},
|
|
{"code": "600519", "name": "贵州茅台"},
|
|
{"code": "000002", "name": "万科A"},
|
|
]
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# Symbol validation
|
|
# ============================================================
|
|
|
|
|
|
class TestValidateAShare:
|
|
def test_valid_starts_with_0(self):
|
|
assert akshare_service.validate_a_share_symbol("000001") is True
|
|
|
|
def test_valid_starts_with_3(self):
|
|
assert akshare_service.validate_a_share_symbol("300001") is True
|
|
|
|
def test_valid_starts_with_6(self):
|
|
assert akshare_service.validate_a_share_symbol("600519") is True
|
|
|
|
def test_invalid_starts_with_1(self):
|
|
assert akshare_service.validate_a_share_symbol("100001") is False
|
|
|
|
def test_invalid_too_short(self):
|
|
assert akshare_service.validate_a_share_symbol("00001") is False
|
|
|
|
def test_invalid_too_long(self):
|
|
assert akshare_service.validate_a_share_symbol("0000011") is False
|
|
|
|
def test_invalid_letters(self):
|
|
assert akshare_service.validate_a_share_symbol("00000A") is False
|
|
|
|
def test_invalid_empty(self):
|
|
assert akshare_service.validate_a_share_symbol("") is False
|
|
|
|
|
|
class TestValidateHKSymbol:
|
|
def test_valid_five_digits(self):
|
|
assert akshare_service.validate_hk_symbol("00700") is True
|
|
|
|
def test_valid_all_nines(self):
|
|
assert akshare_service.validate_hk_symbol("99999") is True
|
|
|
|
def test_invalid_too_short(self):
|
|
assert akshare_service.validate_hk_symbol("0070") is False
|
|
|
|
def test_invalid_too_long(self):
|
|
assert akshare_service.validate_hk_symbol("007000") is False
|
|
|
|
def test_invalid_letters(self):
|
|
assert akshare_service.validate_hk_symbol("0070A") is False
|
|
|
|
def test_invalid_empty(self):
|
|
assert akshare_service.validate_hk_symbol("") is False
|
|
|
|
|
|
# ============================================================
|
|
# _parse_hist_df
|
|
# ============================================================
|
|
|
|
|
|
class TestParseHistDf:
|
|
def test_returns_list_of_dicts(self):
|
|
df = _make_hist_df(2)
|
|
result = akshare_service._parse_hist_df(df)
|
|
assert isinstance(result, list)
|
|
assert len(result) == 2
|
|
|
|
def test_keys_are_english(self):
|
|
df = _make_hist_df(1)
|
|
result = akshare_service._parse_hist_df(df)
|
|
row = result[0]
|
|
assert "date" in row
|
|
assert "open" in row
|
|
assert "close" in row
|
|
assert "high" in row
|
|
assert "low" in row
|
|
assert "volume" in row
|
|
assert "turnover" in row
|
|
assert "change_percent" in row
|
|
assert "turnover_rate" in row
|
|
|
|
def test_no_chinese_keys_remain(self):
|
|
df = _make_hist_df(1)
|
|
result = akshare_service._parse_hist_df(df)
|
|
row = result[0]
|
|
for key in row:
|
|
assert not any(ord(c) > 127 for c in key), f"Non-ASCII key found: {key}"
|
|
|
|
def test_date_is_string(self):
|
|
df = _make_hist_df(1)
|
|
result = akshare_service._parse_hist_df(df)
|
|
assert isinstance(result[0]["date"], str)
|
|
|
|
def test_values_are_correct(self):
|
|
df = _make_hist_df(1)
|
|
result = akshare_service._parse_hist_df(df)
|
|
assert result[0]["open"] == pytest.approx(10.0)
|
|
assert result[0]["close"] == pytest.approx(10.5)
|
|
|
|
def test_empty_df_returns_empty_list(self):
|
|
df = pd.DataFrame()
|
|
result = akshare_service._parse_hist_df(df)
|
|
assert result == []
|
|
|
|
|
|
# ============================================================
|
|
# _parse_spot_row
|
|
# ============================================================
|
|
|
|
|
|
class TestParseSpotRow:
|
|
def test_returns_dict_with_english_keys(self):
|
|
df = _make_spot_df("000001", "平安银行")
|
|
result = akshare_service._parse_spot_row(df, "000001")
|
|
assert result is not None
|
|
assert "symbol" in result
|
|
assert "name" in result
|
|
assert "price" in result
|
|
|
|
def test_correct_symbol_extracted(self):
|
|
df = _make_spot_df("000001")
|
|
result = akshare_service._parse_spot_row(df, "000001")
|
|
assert result["symbol"] == "000001"
|
|
|
|
def test_returns_none_when_symbol_not_found(self):
|
|
df = _make_spot_df("000001")
|
|
result = akshare_service._parse_spot_row(df, "999999")
|
|
assert result is None
|
|
|
|
def test_price_value_correct(self):
|
|
df = _make_spot_df("600519")
|
|
df["代码"] = "600519"
|
|
result = akshare_service._parse_spot_row(df, "600519")
|
|
assert result["price"] == pytest.approx(12.34)
|
|
|
|
def test_all_quote_fields_present(self):
|
|
df = _make_spot_df("000001")
|
|
result = akshare_service._parse_spot_row(df, "000001")
|
|
expected_keys = {
|
|
"symbol", "name", "price", "change", "change_percent",
|
|
"volume", "turnover", "open", "high", "low", "prev_close",
|
|
}
|
|
assert expected_keys.issubset(set(result.keys()))
|
|
|
|
|
|
# ============================================================
|
|
# get_a_share_quote
|
|
# ============================================================
|
|
|
|
|
|
class TestGetAShareQuote:
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_spot_em")
|
|
async def test_returns_quote_dict(self, mock_spot):
|
|
mock_spot.return_value = _make_spot_df("000001")
|
|
result = await akshare_service.get_a_share_quote("000001")
|
|
assert result is not None
|
|
assert result["symbol"] == "000001"
|
|
assert result["name"] == "平安银行"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_spot_em")
|
|
async def test_returns_none_for_unknown_symbol(self, mock_spot):
|
|
mock_spot.return_value = _make_spot_df("000001")
|
|
result = await akshare_service.get_a_share_quote("999999")
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_spot_em")
|
|
async def test_propagates_exception(self, mock_spot):
|
|
mock_spot.side_effect = RuntimeError("AKShare unavailable")
|
|
with pytest.raises(RuntimeError):
|
|
await akshare_service.get_a_share_quote("000001")
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_spot_em")
|
|
async def test_akshare_called_once(self, mock_spot):
|
|
mock_spot.return_value = _make_spot_df("000001")
|
|
await akshare_service.get_a_share_quote("000001")
|
|
mock_spot.assert_called_once()
|
|
|
|
|
|
# ============================================================
|
|
# get_a_share_historical
|
|
# ============================================================
|
|
|
|
|
|
class TestGetAShareHistorical:
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_hist")
|
|
async def test_returns_list_of_bars(self, mock_hist):
|
|
mock_hist.return_value = _make_hist_df(5)
|
|
result = await akshare_service.get_a_share_historical("000001", days=30)
|
|
assert isinstance(result, list)
|
|
assert len(result) == 5
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_hist")
|
|
async def test_bars_have_english_keys(self, mock_hist):
|
|
mock_hist.return_value = _make_hist_df(1)
|
|
result = await akshare_service.get_a_share_historical("000001", days=30)
|
|
assert "date" in result[0]
|
|
assert "open" in result[0]
|
|
assert "close" in result[0]
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_hist")
|
|
async def test_called_with_correct_symbol(self, mock_hist):
|
|
mock_hist.return_value = _make_hist_df(1)
|
|
await akshare_service.get_a_share_historical("600519", days=90)
|
|
call_kwargs = mock_hist.call_args
|
|
assert call_kwargs.kwargs.get("symbol") == "600519" or call_kwargs.args[0] == "600519"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_hist")
|
|
async def test_adjust_is_qfq(self, mock_hist):
|
|
mock_hist.return_value = _make_hist_df(1)
|
|
await akshare_service.get_a_share_historical("000001", days=30)
|
|
call_kwargs = mock_hist.call_args
|
|
assert call_kwargs.kwargs.get("adjust") == "qfq"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_hist")
|
|
async def test_empty_df_returns_empty_list(self, mock_hist):
|
|
mock_hist.return_value = pd.DataFrame()
|
|
result = await akshare_service.get_a_share_historical("000001", days=30)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_zh_a_hist")
|
|
async def test_propagates_exception(self, mock_hist):
|
|
mock_hist.side_effect = RuntimeError("network error")
|
|
with pytest.raises(RuntimeError):
|
|
await akshare_service.get_a_share_historical("000001", days=30)
|
|
|
|
|
|
# ============================================================
|
|
# search_a_shares
|
|
# ============================================================
|
|
|
|
|
|
class TestSearchAShares:
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_info_a_code_name")
|
|
async def test_returns_matching_results(self, mock_codes):
|
|
mock_codes.return_value = _make_code_name_df()
|
|
result = await akshare_service.search_a_shares("平安")
|
|
assert len(result) == 1
|
|
assert result[0]["code"] == "000001"
|
|
assert result[0]["name"] == "平安银行"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_info_a_code_name")
|
|
async def test_returns_empty_list_when_no_match(self, mock_codes):
|
|
mock_codes.return_value = _make_code_name_df()
|
|
result = await akshare_service.search_a_shares("NONEXISTENT")
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_info_a_code_name")
|
|
async def test_returns_multiple_matches(self, mock_codes):
|
|
mock_codes.return_value = _make_code_name_df()
|
|
result = await akshare_service.search_a_shares("万")
|
|
assert len(result) == 1
|
|
assert result[0]["code"] == "000002"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_info_a_code_name")
|
|
async def test_result_has_code_and_name_keys(self, mock_codes):
|
|
mock_codes.return_value = _make_code_name_df()
|
|
result = await akshare_service.search_a_shares("茅台")
|
|
assert len(result) == 1
|
|
assert "code" in result[0]
|
|
assert "name" in result[0]
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_info_a_code_name")
|
|
async def test_propagates_exception(self, mock_codes):
|
|
mock_codes.side_effect = RuntimeError("timeout")
|
|
with pytest.raises(RuntimeError):
|
|
await akshare_service.search_a_shares("平安")
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_info_a_code_name")
|
|
async def test_empty_query_returns_all(self, mock_codes):
|
|
mock_codes.return_value = _make_code_name_df()
|
|
result = await akshare_service.search_a_shares("")
|
|
assert len(result) == 3
|
|
|
|
|
|
# ============================================================
|
|
# get_hk_quote
|
|
# ============================================================
|
|
|
|
|
|
class TestGetHKQuote:
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_spot_em")
|
|
async def test_returns_quote_dict(self, mock_spot):
|
|
mock_spot.return_value = _make_spot_df("00700", "腾讯控股")
|
|
result = await akshare_service.get_hk_quote("00700")
|
|
assert result is not None
|
|
assert result["symbol"] == "00700"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_spot_em")
|
|
async def test_returns_none_for_unknown_symbol(self, mock_spot):
|
|
mock_spot.return_value = _make_spot_df("00700")
|
|
result = await akshare_service.get_hk_quote("99999")
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_spot_em")
|
|
async def test_propagates_exception(self, mock_spot):
|
|
mock_spot.side_effect = RuntimeError("AKShare unavailable")
|
|
with pytest.raises(RuntimeError):
|
|
await akshare_service.get_hk_quote("00700")
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_spot_em")
|
|
async def test_all_fields_present(self, mock_spot):
|
|
mock_spot.return_value = _make_spot_df("00700", "腾讯控股")
|
|
result = await akshare_service.get_hk_quote("00700")
|
|
expected_keys = {
|
|
"symbol", "name", "price", "change", "change_percent",
|
|
"volume", "turnover", "open", "high", "low", "prev_close",
|
|
}
|
|
assert expected_keys.issubset(set(result.keys()))
|
|
|
|
|
|
# ============================================================
|
|
# get_hk_historical
|
|
# ============================================================
|
|
|
|
|
|
class TestGetHKHistorical:
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_hist")
|
|
async def test_returns_list_of_bars(self, mock_hist):
|
|
mock_hist.return_value = _make_hist_df(4)
|
|
result = await akshare_service.get_hk_historical("00700", days=30)
|
|
assert isinstance(result, list)
|
|
assert len(result) == 4
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_hist")
|
|
async def test_bars_have_english_keys(self, mock_hist):
|
|
mock_hist.return_value = _make_hist_df(1)
|
|
result = await akshare_service.get_hk_historical("00700", days=30)
|
|
assert "date" in result[0]
|
|
assert "close" in result[0]
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_hist")
|
|
async def test_called_with_correct_symbol(self, mock_hist):
|
|
mock_hist.return_value = _make_hist_df(1)
|
|
await akshare_service.get_hk_historical("09988", days=90)
|
|
call_kwargs = mock_hist.call_args
|
|
assert call_kwargs.kwargs.get("symbol") == "09988" or call_kwargs.args[0] == "09988"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_hist")
|
|
async def test_adjust_is_qfq(self, mock_hist):
|
|
mock_hist.return_value = _make_hist_df(1)
|
|
await akshare_service.get_hk_historical("00700", days=30)
|
|
call_kwargs = mock_hist.call_args
|
|
assert call_kwargs.kwargs.get("adjust") == "qfq"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_hist")
|
|
async def test_empty_df_returns_empty_list(self, mock_hist):
|
|
mock_hist.return_value = pd.DataFrame()
|
|
result = await akshare_service.get_hk_historical("00700", days=30)
|
|
assert result == []
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("akshare_service.ak.stock_hk_hist")
|
|
async def test_propagates_exception(self, mock_hist):
|
|
mock_hist.side_effect = RuntimeError("network error")
|
|
with pytest.raises(RuntimeError):
|
|
await akshare_service.get_hk_historical("00700", days=30)
|