toor
11ddbd3c49
- 新增 9 个抓取模块: 三大报表/盈利能力/资本结构/营运能力/成长能力/杜邦/分红 - api.py 新增 10 个 /fin/* 接口并统一无数据返回 - 新增 _emchoice.py 共享请求辅助 (emws 跳转取 URL / datacenter 报表拉取) - 新增冒烟测试 tests/test_interfaces.py (24 用例) - 新增接口文档 api_manual.md 与施工蓝图 blueprint.md - 新增 Dockerfile / docker-compose.yml / deploy.sh 部署配置
85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
"""接口冒烟测试: 直接调用 utils 各模块函数, 连真实东方财富接口验证能取到数据.
|
|
|
|
运行: uv run pytest tests/ -q
|
|
注意: 依赖外网可达东财接口, 网络不通时用例会失败.
|
|
"""
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from utils.fin_balance import get_balance
|
|
from utils.fin_capital_structure import get_capital_structure
|
|
from utils.fin_cashflow import get_cashflow
|
|
from utils.fin_dividend import get_dividend_detail, get_dividend_statics
|
|
from utils.fin_dupont import get_dupont
|
|
from utils.fin_growth import get_growth
|
|
from utils.fin_income import get_income
|
|
from utils.fin_main_business import get_main_business
|
|
from utils.fin_operate import get_operate_ability
|
|
from utils.fin_profitability import get_profitability
|
|
|
|
STOCK = "603233.SH"
|
|
|
|
# (模块函数, 该接口应返回的关键列, 测试名)
|
|
CASES = [
|
|
pytest.param(get_main_business, "MAIN_BUSINESS_INCOME", id="main_business"),
|
|
pytest.param(get_balance, "TOTAL_ASSETS", id="balance"),
|
|
pytest.param(get_income, "TOTAL_OPERATE_INCOME", id="income"),
|
|
pytest.param(get_cashflow, "SALES_SERVICES", id="cashflow"),
|
|
pytest.param(get_profitability, "ROE_AVERAGE", id="profitability"),
|
|
pytest.param(get_capital_structure, "DEBT_ASSET_RATIO", id="capital_structure"),
|
|
pytest.param(get_operate_ability, "OPERATE_CYCLE", id="operate_ability"),
|
|
pytest.param(get_growth, "BASICEPS_YOY", id="growth"),
|
|
pytest.param(get_dupont, "ROE", id="dupont"),
|
|
pytest.param(get_dividend_detail, "IMPL_PLAN_PROFILE", id="dividend_detail"),
|
|
pytest.param(get_dividend_statics, "AUALACCMDIV_ARD", id="dividend_statics"),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("func,key_col", CASES)
|
|
def test_fetch_data(func, key_col):
|
|
"""拉 1 年数据, 应返回非空 DataFrame 且含关键列."""
|
|
df = func([STOCK], years=1)
|
|
assert isinstance(df, pd.DataFrame)
|
|
assert not df.empty, f"{func.__name__} 未返回数据"
|
|
assert "SECUCODE" in df.columns, f"{func.__name__} 缺少 SECUCODE 列"
|
|
assert key_col in df.columns, f"{func.__name__} 缺少关键列 {key_col}"
|
|
|
|
|
|
@pytest.mark.parametrize("func,_key_col", CASES)
|
|
def test_empty_codes(func, _key_col):
|
|
"""空代码列表应返回空 DataFrame, 不抛异常."""
|
|
df = func([], years=1)
|
|
assert isinstance(df, pd.DataFrame)
|
|
assert df.empty
|
|
|
|
|
|
def test_main_business_region():
|
|
"""主营构成按地区分类也能取到数据."""
|
|
df = get_main_business([STOCK], classify_type="地区", years=1)
|
|
assert not df.empty
|
|
assert df["CLASSIFY_TYPE"].iloc[0] == "地区"
|
|
|
|
|
|
def test_api_routes():
|
|
"""api.py 的所有 HTTP 路由已注册."""
|
|
import api
|
|
|
|
paths = {r.path for r in api.app.routes if hasattr(r, "path")}
|
|
for route in [
|
|
"/health",
|
|
"/fin/summary",
|
|
"/fin/quarterly-summary",
|
|
"/fin/main-business",
|
|
"/fin/balance",
|
|
"/fin/income",
|
|
"/fin/cashflow",
|
|
"/fin/profitability",
|
|
"/fin/capital-structure",
|
|
"/fin/operate-ability",
|
|
"/fin/growth",
|
|
"/fin/dupont",
|
|
"/fin/dividend-detail",
|
|
"/fin/dividend-statics",
|
|
]:
|
|
assert route in paths, f"缺少路由 {route}"
|