ad70d8fccc
- New provider_call_log/provider_health tables (additive-only schema), wired via a fail-open observability.observe()/record_call() helper. - Tushare pipeline keeps its existing src_calls record unchanged and now also feeds the unified provider_health/provider_call_log side channel. - Eastmoney/Tencent realtime_serve.py call sites and the iFinD steward call site are wrapped with observability.observe() at the call site only; no adapter internals, routing, fallback order, or return values are touched. - New read-only admin API endpoints: /admin/api/providers/status, /admin/api/source-catalog, /admin/api/lineage, /admin/api/lineage/affected. - New static, read-only source_catalog.py and lineage.py registries documenting existing providers/interfaces/datasets and known main-site consumers (cited against backend/features/screener and backend/features/heaven call sites). - provider_call_log is purged by the existing pipeline.cleanup() job alongside src_calls/job_runs. - 47 new unit/integration tests covering classification, fail-open behavior under DB/log failures, unchanged payloads/exceptions on success and failure paths, and the new HTTP endpoints. Full suite: 173 tests, all green. Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from datahub.adapters.ifind import IfindAdapter
|
|
from datahub.db import HubDB
|
|
from datahub.serving import ApiError
|
|
from datahub.steward import steward_query
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, payload: dict, status: int = 200) -> None:
|
|
import json
|
|
|
|
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 IfindObservabilityTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db = HubDB(Path(self.tmp.name) / "hub.db")
|
|
|
|
def tearDown(self) -> None:
|
|
self.tmp.cleanup()
|
|
|
|
def _adapter_with_urlopen(self, urlopen) -> IfindAdapter:
|
|
return IfindAdapter(refresh_token="rt", access_token="at", urlopen=urlopen)
|
|
|
|
def test_successful_fetch_is_logged_without_changing_rows(self) -> None:
|
|
def urlopen(request, timeout=None):
|
|
return _Resp(
|
|
{
|
|
"errorcode": 0,
|
|
"tables": [{"thscode": ["000001.SZ"], "table": {"涨停原因": ["重组"]}}],
|
|
}
|
|
)
|
|
|
|
adapter = self._adapter_with_urlopen(urlopen)
|
|
|
|
class _Api:
|
|
ifind = adapter
|
|
db = self.db
|
|
|
|
payload = steward_query(
|
|
_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}}
|
|
)
|
|
self.assertEqual(payload["data"][0]["thscode"], "000001.SZ")
|
|
log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
|
|
self.assertIsNotNone(log)
|
|
self.assertEqual(log["interface"], "wencai")
|
|
self.assertEqual(log["status"], "ok")
|
|
|
|
def test_failed_fetch_reraises_and_logs_error(self) -> None:
|
|
def urlopen(request, timeout=None):
|
|
return _Resp({"errorcode": -9999, "errmsg": "quota exceeded"})
|
|
|
|
adapter = self._adapter_with_urlopen(urlopen)
|
|
|
|
class _Api:
|
|
ifind = adapter
|
|
db = self.db
|
|
|
|
with self.assertRaises(ApiError) as ctx:
|
|
steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}})
|
|
self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE")
|
|
log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
|
|
self.assertIsNotNone(log)
|
|
self.assertEqual(log["status"], "error")
|
|
|
|
def test_status_check_alone_does_not_dial_or_log_a_fetch_call(self) -> None:
|
|
class _Api:
|
|
ifind = IfindAdapter()
|
|
db = self.db
|
|
|
|
payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}})
|
|
self.assertFalse(payload["data"][0]["configured"])
|
|
log = self.db.fetchall("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
|
|
self.assertEqual(log, [])
|
|
|
|
def test_api_double_without_db_attribute_still_works(self) -> None:
|
|
# Mirrors tests/test_ifind_adapter.py's `_Api` double, which has no
|
|
# `db` attribute at all. Observability must not require it.
|
|
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()
|