"""同花顺公开网页解析及接口测试。"""
from __future__ import annotations
import httpx
import pytest
pytest.importorskip("fastapi")
from fastapi import FastAPI # noqa: E402
from fastapi.testclient import TestClient # noqa: E402
from easy_tdx.ths_web import ( # noqa: E402
ThsWebClient,
parse_board_detail,
parse_board_directory,
parse_concepts,
parse_industry_hierarchy,
)
from easy_tdx.web.deps import get_ths_web_client # noqa: E402
from easy_tdx.web.routers.ths import router # noqa: E402
_FIELD_HTML = """
三级行业分类：
电子 -- 半导体 -- 数字芯片设计 （共57家）
"""
_CONCEPT_HTML = """
芯片概念
比亚迪概念
"""
_INDUSTRY_DIRECTORY_HTML = """
"""
_CONCEPT_DIRECTORY_HTML = """
"""
_INDUSTRY_DETAIL_HTML = """
"""
_CHIP_DETAIL_HTML = """
"""
_BYD_DETAIL_HTML = """
"""
def test_parse_industry_hierarchy() -> None:
assert parse_industry_hierarchy(_FIELD_HTML) == ["电子", "半导体", "数字芯片设计"]
def test_parse_concepts_extracts_codes_names_and_leading_stock_codes() -> None:
assert parse_concepts(_CONCEPT_HTML) == [
{"board_code": "301085", "name": "芯片概念", "leader_codes": ["002886", "003005"]},
{"board_code": "308972", "name": "比亚迪概念", "leader_codes": ["300750"]},
]
def test_parse_board_directory_and_detail_extract_quote_and_leader() -> None:
assert parse_board_directory(_INDUSTRY_DIRECTORY_HTML, "industry") == {"半导体": "881121"}
assert parse_board_detail(_INDUSTRY_DETAIL_HTML) == {
"board_code": "881121",
"change_pct": 2.35,
"leader": {"code": "688981", "name": "中芯国际", "change_pct": 3.21},
}
@pytest.mark.asyncio
async def test_client_merges_stock_membership_and_board_quotes() -> None:
pages = {
"/603893/field.html": _FIELD_HTML,
"/603893/concept.html": _CONCEPT_HTML,
"/thshy/": _INDUSTRY_DIRECTORY_HTML,
"/gn/": _CONCEPT_DIRECTORY_HTML,
"/thshy/detail/code/881121/": _INDUSTRY_DETAIL_HTML,
"/gn/detail/code/301085/": _CHIP_DETAIL_HTML,
"/gn/detail/code/308972/": _BYD_DETAIL_HTML,
}
calls: dict[str, int] = {}
async def handler(request: httpx.Request) -> httpx.Response:
calls[request.url.path] = calls.get(request.url.path, 0) + 1
return httpx.Response(200, content=pages[request.url.path].encode("gbk"))
client = ThsWebClient(transport=httpx.MockTransport(handler))
result = await client.get_stock_associations("603893")
assert result["code"] == "603893"
assert result["industries"] == [
{"level": 1, "name": "电子", "board_code": None, "change_pct": None, "leader": None},
{
"level": 2,
"name": "半导体",
"board_code": "881121",
"change_pct": 2.35,
"leader": {"code": "688981", "name": "中芯国际", "change_pct": 3.21},
},
{
"level": 3,
"name": "数字芯片设计",
"board_code": None,
"change_pct": None,
"leader": None,
},
]
assert result["concept_total"] == 2
assert result["concepts"] == [
{
"board_code": "301085",
"name": "芯片概念",
"change_pct": -0.05,
"leader": {"code": "002886", "name": "沃特股份", "change_pct": 9.5},
"leader_codes": ["002886", "003005"],
},
{
"board_code": "308972",
"name": "比亚迪概念",
"change_pct": 0.0,
"leader": None,
"leader_codes": ["300750"],
},
]
await client.get_stock_associations("603893")
assert calls == {
"/603893/field.html": 2,
"/603893/concept.html": 2,
"/thshy/": 2,
"/gn/": 2,
"/thshy/detail/code/881121/": 1,
"/gn/detail/code/301085/": 1,
"/gn/detail/code/308972/": 1,
}
class _FakeThsClient:
async def get_stock_associations(self, code: str, concept_limit: int) -> dict[str, object]:
assert concept_limit == 10
return {"source": "ths_web", "code": code, "industries": [], "concepts": []}
def _api_client() -> TestClient:
app = FastAPI()
app.include_router(router, prefix="/api/v1")
app.dependency_overrides[get_ths_web_client] = lambda: _FakeThsClient()
return TestClient(app)
def test_ths_associations_endpoint() -> None:
with _api_client() as client:
response = client.get("/api/v1/ths/stock/associations", params={"code": "603893"})
assert response.status_code == 200
assert response.json() == {
"data": {"source": "ths_web", "code": "603893", "industries": [], "concepts": []}
}
def test_ths_associations_endpoint_validates_code() -> None:
with _api_client() as client:
response = client.get("/api/v1/ths/stock/associations", params={"code": "60389"})
assert response.status_code == 422