34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# 将项目根目录加入 sys.path，确保可以直接导入业务代码
|
|
ROOT_DIR = Path(__file__).resolve().parents[4]
|
|
if str(ROOT_DIR) not in sys.path:
|
|
sys.path.append(str(ROOT_DIR))
|
|
|
|
from user_data.Chan.web.cn_stock_data import ChinaStockData
|
|
|
|
|
|
@pytest.mark.network
|
|
def test_cn_stock_data_fetch():
|
|
"""简单的联通性测试，确认能否通过 akshare 拉取A股K线数据。"""
|
|
data_client = ChinaStockData()
|
|
|
|
try:
|
|
df = data_client.get_kl_data(symbol="600519", timeframe="1d", limit=20)
|
|
except Exception as exc: # pragma: no cover - 旨在提示网络/依赖问题
|
|
pytest.skip(f"无法调用数据接口，可能是网络或依赖问题：{exc}")
|
|
|
|
if df is None or df.empty:
|
|
pytest.skip("未获取到任何数据，可能是网络异常或接口限制。")
|
|
|
|
expected_cols = {"date", "open", "high", "low", "close", "volume"}
|
|
missing_cols = expected_cols - set(df.columns)
|
|
assert not missing_cols, f"返回数据缺少列: {missing_cols}"
|
|
|
|
|