"""Unit tests for akshare_service.py - written FIRST (TDD RED phase).""" from datetime import datetime from unittest.mock import MagicMock, 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)