100752f43c
生产 gateway 不再实例化 iFinD、东财图和免费实时聚合器；问财与竞价快照作为中枢内部数据源。全站阻断外源测试覆盖日K、报价、图表、问财和竞价快照。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
|
|
from datahub.adapters.ifind import IfindAdapter
|
|
from datahub.serving import ApiError
|
|
from datahub.steward import steward_query
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, payload: dict, status: int = 200) -> None:
|
|
self.status = status
|
|
self._raw = json.dumps(payload).encode("utf-8")
|
|
|
|
def read(self):
|
|
return self._raw
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
|
|
class IfindAdapterTests(unittest.TestCase):
|
|
def test_unconfigured_probe_does_not_dial_vendor(self) -> None:
|
|
calls = []
|
|
|
|
def urlopen(request, timeout=None):
|
|
calls.append(str(getattr(request, "full_url", request)))
|
|
raise AssertionError("should not dial")
|
|
|
|
adapter = IfindAdapter(urlopen=urlopen)
|
|
probe = adapter.probe()
|
|
self.assertEqual(probe["state"], "unconfigured")
|
|
self.assertFalse(probe["configured"])
|
|
self.assertEqual(calls, [])
|
|
|
|
def test_wencai_and_snapshots_go_to_ifind_http(self) -> None:
|
|
calls = []
|
|
|
|
def urlopen(request, timeout=None):
|
|
url = str(getattr(request, "full_url", request))
|
|
calls.append(url)
|
|
if url.endswith("get_access_token"):
|
|
return _Resp({"errorcode": 0, "data": {"access_token": "acc", "expired_time": ""}})
|
|
body = json.loads(request.data.decode("utf-8"))
|
|
if url.endswith("smart_stock_picking"):
|
|
self.assertEqual(body["searchstring"], "涨停")
|
|
return _Resp(
|
|
{
|
|
"errorcode": 0,
|
|
"tables": [
|
|
{
|
|
"thscode": ["000001.SZ"],
|
|
"table": {"涨停原因": ["重组"]},
|
|
}
|
|
],
|
|
}
|
|
)
|
|
if url.endswith("snap_shot"):
|
|
return _Resp(
|
|
{
|
|
"errorcode": 0,
|
|
"tables": [
|
|
{
|
|
"time": ["2024-09-02 09:21:00"],
|
|
"thscode": ["000001.SZ"],
|
|
"table": {"latest": [10.5], "preClose": [10]},
|
|
}
|
|
],
|
|
}
|
|
)
|
|
raise AssertionError(url)
|
|
|
|
adapter = IfindAdapter("refresh-token", urlopen=urlopen)
|
|
rows = adapter.fetch("wencai", {"query": "涨停"})
|
|
self.assertEqual(rows[0]["涨停原因"], "重组")
|
|
snaps = adapter.fetch(
|
|
"snapshots",
|
|
{
|
|
"codes": ["000001.SZ"],
|
|
"indicators": ["latest", "preClose"],
|
|
"start_time": "2024-09-02 09:15:00",
|
|
"end_time": "2024-09-02 09:22:00",
|
|
},
|
|
)
|
|
self.assertEqual(snaps[0]["latest"], 10.5)
|
|
self.assertTrue(any("quantapi.51ifind.com" in item for item in calls))
|
|
|
|
def test_steward_ifind_status_and_unconfigured_fetch(self) -> None:
|
|
class _Api:
|
|
ifind = IfindAdapter()
|
|
|
|
payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}})
|
|
self.assertFalse(payload["data"][0]["configured"])
|
|
with self.assertRaises(ApiError):
|
|
steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|