You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
155 lines
5.8 KiB
155 lines
5.8 KiB
"""接口冒烟测试: 直接调用 utils 各模块函数, 连真实东方财富接口验证能取到数据.
|
|
|
|
运行: uv run pytest tests/ -q
|
|
注意: 依赖外网可达东财接口, 网络不通时用例会失败.
|
|
"""
|
|
import inspect
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from utils.fin_balance import get_balance
|
|
from utils.fin_belong_index import get_belong_index, get_index_classif, get_index_membership
|
|
from utils.fin_capital_structure import get_capital_structure
|
|
from utils.fin_cashflow import get_cashflow
|
|
from utils.fin_company import get_company_info, get_industry
|
|
from utils.fin_dividend import get_dividend_detail, get_dividend_statics
|
|
from utils.fin_dupont import get_dupont
|
|
from utils.fin_equity import get_equity
|
|
from utils.fin_free_holders import get_free_holders
|
|
from utils.fin_growth import get_growth
|
|
from utils.fin_holder_num import get_holder_num
|
|
from utils.fin_holders import get_holders
|
|
from utils.fin_income import get_income
|
|
from utils.fin_main_business import get_main_business
|
|
from utils.fin_major_events import (
|
|
get_acquisitions,
|
|
get_equity_incentive,
|
|
get_guarantee,
|
|
get_litigation,
|
|
get_related_transaction,
|
|
get_violation,
|
|
)
|
|
from utils.fin_mda import get_mda
|
|
from utils.fin_operate import get_operate_ability
|
|
from utils.fin_profitability import get_profitability
|
|
|
|
STOCK = "603233.SH"
|
|
|
|
# 部分接口需要更多年份或其它股票才能取到数据 (否则按 1 年 + 默认股票)
|
|
CASE_OVERRIDES = {
|
|
"get_mda": {"years": 5},
|
|
"get_litigation": {"years": 3},
|
|
"get_violation": {"years": 3},
|
|
"get_equity_incentive": {"years": 5, "stock": "300750.SZ"},
|
|
"get_guarantee": {"years": 3, "stock": "300750.SZ"},
|
|
}
|
|
|
|
|
|
def _call(func, stock=STOCK):
|
|
"""按函数签名与覆盖配置调用."""
|
|
over = CASE_OVERRIDES.get(func.__name__, {})
|
|
s = over.get("stock", stock)
|
|
if "years" in inspect.signature(func).parameters:
|
|
return func([s], years=over.get("years", 1))
|
|
return func([s])
|
|
|
|
|
|
# (模块函数, 该接口应返回的关键列, 测试名)
|
|
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.param(get_free_holders, "HOLDER_NAME", id="free_holders"),
|
|
pytest.param(get_holders, "HOLDER_NAME", id="holders"),
|
|
pytest.param(get_holder_num, "HOLDER_TOTAL_NUM", id="holder_num"),
|
|
pytest.param(get_mda, "REPORT_NAME", id="mda"),
|
|
pytest.param(get_equity, "TOTAL_SHARES", id="equity"),
|
|
pytest.param(get_company_info, "ORG_NAME", id="company_info"),
|
|
pytest.param(get_industry, "INDUSTRY_NAME", id="industry"),
|
|
pytest.param(get_index_classif, "INDEX_CLASSIF", id="index_classif"),
|
|
pytest.param(get_belong_index, "INDEX_NUM", id="belong_index"),
|
|
pytest.param(get_index_membership, "INDEX_NAME_ABBR", id="index_membership"),
|
|
pytest.param(get_acquisitions, "PLAN_PROCESS", id="acquisitions"),
|
|
pytest.param(get_equity_incentive, "PLAN_PROCESS", id="equity_incentive"),
|
|
pytest.param(get_litigation, "CASE_NAME", id="litigation"),
|
|
pytest.param(get_related_transaction, "RELATED_PARTY", id="related_transaction"),
|
|
pytest.param(get_violation, "VIOLATE_TYPE", id="violation"),
|
|
pytest.param(get_guarantee, "GUAR_NAME", id="guarantee"),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("func,key_col", CASES)
|
|
def test_fetch_data(func, key_col):
|
|
"""拉数据, 应返回非空 DataFrame 且含关键列."""
|
|
df = _call(func)
|
|
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, 不抛异常."""
|
|
if "years" in inspect.signature(func).parameters:
|
|
df = func([], years=1)
|
|
else:
|
|
df = func([])
|
|
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",
|
|
"/fin/free-holders",
|
|
"/fin/holders",
|
|
"/fin/holder-num",
|
|
"/fin/mda",
|
|
"/fin/equity",
|
|
"/fin/company-info",
|
|
"/fin/industry",
|
|
"/fin/index-classif",
|
|
"/fin/belong-index",
|
|
"/fin/index-membership",
|
|
"/fin/acquisitions",
|
|
"/fin/equity-incentive",
|
|
"/fin/litigation",
|
|
"/fin/related-transaction",
|
|
"/fin/violation",
|
|
"/fin/guarantee",
|
|
]:
|
|
assert route in paths, f"缺少路由 {route}"
|
|
|