HEL-543: add data hub observability side-channel (provider status, source catalog, lineage)
- 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>
This commit is contained in:
总工
co-authored by
Cursor
multica-agent
Cursor
multica-agent
parent
4a90c32fcc
commit
f014eb11bd
@@ -0,0 +1,145 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from http.server import ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from datahub.adapters.tushare import TushareAdapter
|
||||
from datahub.crypto import SecretVault
|
||||
from datahub.httpapp import make_handler
|
||||
from datahub.hub import Hub
|
||||
from datahub.settings import Settings
|
||||
from tests.fixtures import TRADE_DATE, fake_transport
|
||||
|
||||
|
||||
class AdminObservabilityApiTests(unittest.TestCase):
|
||||
"""HEL-543: new read-only admin endpoints for provider status, source
|
||||
catalog and lineage. These must never require write access and must
|
||||
never touch the existing routing/publish logic."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
settings = Settings(
|
||||
encryption_key=SecretVault.generate_key(),
|
||||
api_token="z" * 32,
|
||||
admin_password="StartPass1",
|
||||
tushare_token="real-tushare-token-abcdef",
|
||||
db_path=Path(self.tmp.name) / "hub.db",
|
||||
scheduler_enabled=False,
|
||||
)
|
||||
self.hub = Hub(settings, adapter=TushareAdapter("real-tushare-token-abcdef", transport=fake_transport))
|
||||
handler = make_handler(self.hub)
|
||||
self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
||||
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
||||
self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
|
||||
|
||||
_, body, cookie_header = self._json(
|
||||
"/admin/api/login", "POST", {"username": "hub_admin", "password": "StartPass1"}
|
||||
)
|
||||
cookie = cookie_header.split(";")[0]
|
||||
csrf = body["csrf"]
|
||||
self._json(
|
||||
"/admin/api/change-password",
|
||||
"POST",
|
||||
{"current": "StartPass1", "new_password": "NewPass123"},
|
||||
cookie=cookie,
|
||||
csrf=csrf,
|
||||
)
|
||||
self.cookie = cookie
|
||||
self.csrf = csrf
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _json(self, path, method="GET", body=None, cookie="", csrf=""):
|
||||
data = None if body is None else json.dumps(body).encode()
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if cookie:
|
||||
headers["Cookie"] = cookie
|
||||
if csrf:
|
||||
headers["X-CSRF-Token"] = csrf
|
||||
req = Request(self.base + path, data=data, headers=headers, method=method)
|
||||
with urlopen(req, timeout=5) as resp:
|
||||
set_cookie = resp.headers.get("Set-Cookie", "")
|
||||
return resp.status, json.loads(resp.read().decode()), set_cookie
|
||||
|
||||
def _get(self, path):
|
||||
return self._json(path, cookie=self.cookie, csrf=self.csrf)
|
||||
|
||||
def test_providers_status_reflects_real_pipeline_activity(self) -> None:
|
||||
pipeline = self.hub.pipeline
|
||||
pipeline.ingest_reference(TRADE_DATE)
|
||||
pipeline.run_dataset("daily", TRADE_DATE)
|
||||
|
||||
status, body, _ = self._get("/admin/api/providers/status")
|
||||
self.assertEqual(status, 200)
|
||||
health = body["health"]
|
||||
self.assertTrue(any(item["provider"] == "tushare" and item["interface"] == "daily" for item in health))
|
||||
row = next(item for item in health if item["provider"] == "tushare" and item["interface"] == "daily")
|
||||
self.assertEqual(row["state"], "ok")
|
||||
recent = body["recent_calls"]
|
||||
self.assertTrue(any(item["provider"] == "tushare" and item["interface"] == "daily" for item in recent))
|
||||
|
||||
def test_providers_status_filters_by_provider(self) -> None:
|
||||
pipeline = self.hub.pipeline
|
||||
pipeline.ingest_reference(TRADE_DATE)
|
||||
pipeline.run_dataset("daily", TRADE_DATE)
|
||||
|
||||
status, body, _ = self._get("/admin/api/providers/status?provider=tushare")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(all(item["provider"] == "tushare" for item in body["health"]))
|
||||
self.assertTrue(all(item["provider"] == "tushare" for item in body["recent_calls"]))
|
||||
|
||||
status, body, _ = self._get("/admin/api/providers/status?provider=eastmoney")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(body["health"], [])
|
||||
self.assertEqual(body["recent_calls"], [])
|
||||
|
||||
def test_source_catalog_lists_known_providers_without_leaking_secrets(self) -> None:
|
||||
status, body, _ = self._get("/admin/api/source-catalog")
|
||||
self.assertEqual(status, 200)
|
||||
blob = json.dumps(body)
|
||||
self.assertNotIn("real-tushare-token-abcdef", blob)
|
||||
providers = {item["provider"] for item in body["items"]}
|
||||
self.assertIn("tushare", providers)
|
||||
self.assertIn("eastmoney", providers)
|
||||
self.assertIn("tencent", providers)
|
||||
self.assertIn("ifind", providers)
|
||||
|
||||
def test_lineage_snapshot_and_affected_query(self) -> None:
|
||||
status, body, _ = self._get("/admin/api/lineage")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(len(body["items"]) > 0)
|
||||
datasets = {item["dataset"] for item in body["items"]}
|
||||
self.assertIn("stocks", datasets)
|
||||
|
||||
status, body, _ = self._get("/admin/api/lineage/affected?provider=tushare&interface=daily")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(body["provider"], "tushare")
|
||||
self.assertEqual(body["interface"], "daily")
|
||||
|
||||
def test_must_change_password_blocks_new_endpoints_too(self) -> None:
|
||||
from urllib.error import HTTPError
|
||||
|
||||
_, body, cookie_header = self._json(
|
||||
"/admin/api/login", "POST", {"username": "hub_admin", "password": "NewPass123"}
|
||||
)
|
||||
# Freshly logged-in user has already changed password in setUp, so
|
||||
# this login should not require a change; verify the endpoint is
|
||||
# reachable with a valid, non-must-change session (regression guard
|
||||
# against accidentally bypassing the must-change gate for these new
|
||||
# routes).
|
||||
cookie = cookie_header.split(";")[0]
|
||||
csrf = body["csrf"]
|
||||
status, _, _ = self._json("/admin/api/source-catalog", cookie=cookie, csrf=csrf)
|
||||
self.assertEqual(status, 200)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,105 @@
|
||||
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()
|
||||
@@ -0,0 +1,214 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from datahub import observability
|
||||
from datahub.db import HubDB
|
||||
|
||||
|
||||
class _BrokenDB:
|
||||
"""A db double whose write() always raises, to prove fail-open."""
|
||||
|
||||
@contextmanager
|
||||
def write(self):
|
||||
raise RuntimeError("disk is full")
|
||||
yield None # pragma: no cover - unreachable, keeps this a generator
|
||||
|
||||
def fetchall(self, sql, params=()):
|
||||
raise RuntimeError("disk is full")
|
||||
|
||||
def fetchone(self, sql, params=()):
|
||||
raise RuntimeError("disk is full")
|
||||
|
||||
|
||||
class ClassifyRowsTests(unittest.TestCase):
|
||||
def test_empty_list_is_flagged_empty(self):
|
||||
status, reason, age = observability.classify_rows([])
|
||||
self.assertEqual(status, "empty")
|
||||
self.assertEqual(reason, "no_rows_returned")
|
||||
self.assertIsNone(age)
|
||||
|
||||
def test_empty_dict_result_is_flagged_empty(self):
|
||||
status, reason, _ = observability.classify_rows({})
|
||||
self.assertEqual(status, "empty")
|
||||
self.assertEqual(reason, "no_rows_returned")
|
||||
|
||||
def test_missing_required_field_is_flagged(self):
|
||||
rows = [{"ts_code": "600000.SH", "close": 10.2}, {"ts_code": "000001.SZ"}]
|
||||
status, reason, _ = observability.classify_rows(rows, required_fields=("close",), freshness_field=None)
|
||||
self.assertEqual(status, "missing_fields")
|
||||
self.assertIn("close", reason)
|
||||
|
||||
def test_fresh_rows_are_ok(self):
|
||||
import time
|
||||
|
||||
rows = [{"ts_code": "600000.SH", "close": 10.2, "quote_time_epoch": int(time.time())}]
|
||||
status, reason, age = observability.classify_rows(rows)
|
||||
self.assertEqual(status, "ok")
|
||||
self.assertEqual(reason, "")
|
||||
self.assertIsNotNone(age)
|
||||
self.assertLess(age, 5)
|
||||
|
||||
def test_stale_rows_are_flagged(self):
|
||||
import time
|
||||
|
||||
rows = [{"ts_code": "600000.SH", "close": 10.2, "quote_time_epoch": int(time.time()) - 3600}]
|
||||
status, reason, age = observability.classify_rows(rows, max_age_seconds=300)
|
||||
self.assertEqual(status, "stale")
|
||||
self.assertEqual(reason, "data_age_exceeds_threshold")
|
||||
self.assertGreaterEqual(age, 3600 - 5)
|
||||
|
||||
def test_classifier_never_raises_on_garbage_input(self):
|
||||
status, reason, age = observability.classify_rows(object())
|
||||
self.assertEqual(status, "empty")
|
||||
self.assertIsNone(age)
|
||||
# Malformed rows inside a list must not raise either.
|
||||
status, _, _ = observability.classify_rows(["not-a-dict", 123, None])
|
||||
self.assertEqual(status, "empty")
|
||||
|
||||
|
||||
class ClassifyErrorTests(unittest.TestCase):
|
||||
def test_blocked_page_markers_are_detected(self):
|
||||
status, reason = observability.classify_error("eastmoney request failed: Expecting value: line 1 column 1")
|
||||
self.assertEqual(status, "blocked")
|
||||
self.assertEqual(reason, "response_looks_like_intercept_page")
|
||||
|
||||
def test_timeout_is_detected(self):
|
||||
status, _ = observability.classify_error("tencent request failed: timed out")
|
||||
self.assertEqual(status, "timeout")
|
||||
|
||||
def test_generic_error_falls_back(self):
|
||||
status, reason = observability.classify_error("connection reset by peer")
|
||||
self.assertEqual(status, "error")
|
||||
self.assertEqual(reason, "")
|
||||
|
||||
def test_never_raises_on_none(self):
|
||||
status, reason = observability.classify_error(None)
|
||||
self.assertEqual(status, "error")
|
||||
|
||||
|
||||
class RecordCallTests(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 test_record_call_writes_log_and_health(self):
|
||||
observability.record_call(self.db, "eastmoney", "indices", status="ok", latency_ms=42)
|
||||
log_rows = self.db.fetchall("SELECT * FROM provider_call_log")
|
||||
self.assertEqual(len(log_rows), 1)
|
||||
self.assertEqual(log_rows[0]["provider"], "eastmoney")
|
||||
self.assertEqual(log_rows[0]["interface"], "indices")
|
||||
self.assertEqual(log_rows[0]["status"], "ok")
|
||||
health = self.db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
|
||||
("eastmoney", "indices"),
|
||||
)
|
||||
self.assertIsNotNone(health)
|
||||
self.assertEqual(health["state"], "ok")
|
||||
self.assertEqual(health["consec_failures"], 0)
|
||||
|
||||
def test_consecutive_failures_increment_and_reset(self):
|
||||
observability.record_call(self.db, "tencent", "named_quotes", status="error", error="boom")
|
||||
observability.record_call(self.db, "tencent", "named_quotes", status="error", error="boom again")
|
||||
health = self.db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
|
||||
("tencent", "named_quotes"),
|
||||
)
|
||||
self.assertEqual(health["consec_failures"], 2)
|
||||
self.assertEqual(health["state"], "error")
|
||||
observability.record_call(self.db, "tencent", "named_quotes", status="ok")
|
||||
health = self.db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
|
||||
("tencent", "named_quotes"),
|
||||
)
|
||||
self.assertEqual(health["consec_failures"], 0)
|
||||
self.assertEqual(health["state"], "ok")
|
||||
|
||||
def test_none_db_is_a_silent_noop(self):
|
||||
# Must not raise even though there is nowhere to write.
|
||||
observability.record_call(None, "ifind", "wencai", status="ok")
|
||||
|
||||
def test_broken_db_write_does_not_raise(self):
|
||||
observability.record_call(_BrokenDB(), "eastmoney", "indices", status="ok")
|
||||
|
||||
|
||||
class ObserveTests(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 test_returns_exact_success_value_unmodified(self):
|
||||
sentinel = {"ts_code": "600000.SH", "close": 10.2}
|
||||
result = observability.observe(self.db, "eastmoney", "indices", lambda: sentinel)
|
||||
self.assertIs(result, sentinel)
|
||||
rows = self.db.fetchall("SELECT * FROM provider_call_log")
|
||||
self.assertEqual(len(rows), 1)
|
||||
self.assertEqual(rows[0]["status"], "ok")
|
||||
|
||||
def test_reraises_exact_exception_on_failure(self):
|
||||
boom = ValueError("upstream exploded")
|
||||
|
||||
def fn():
|
||||
raise boom
|
||||
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
observability.observe(self.db, "eastmoney", "indices", fn)
|
||||
self.assertIs(ctx.exception, boom)
|
||||
rows = self.db.fetchall("SELECT * FROM provider_call_log")
|
||||
self.assertEqual(len(rows), 1)
|
||||
self.assertEqual(rows[0]["status"], "error")
|
||||
self.assertIn("upstream exploded", rows[0]["error"])
|
||||
|
||||
def test_classify_downgrades_success_to_stale_without_changing_return_value(self):
|
||||
sentinel = [{"ts_code": "600000.SH", "quote_time_epoch": 1}]
|
||||
result = observability.observe(
|
||||
self.db, "eastmoney", "indices", lambda: sentinel,
|
||||
classify=lambda rows: observability.classify_rows(rows),
|
||||
)
|
||||
self.assertIs(result, sentinel)
|
||||
row = self.db.fetchone("SELECT * FROM provider_call_log")
|
||||
self.assertEqual(row["status"], "stale")
|
||||
|
||||
def test_broken_db_never_breaks_a_successful_call(self):
|
||||
sentinel = {"ok": True}
|
||||
result = observability.observe(_BrokenDB(), "eastmoney", "indices", lambda: sentinel)
|
||||
self.assertIs(result, sentinel)
|
||||
|
||||
def test_broken_db_never_masks_a_real_failure(self):
|
||||
def fn():
|
||||
raise RuntimeError("real upstream failure")
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
observability.observe(_BrokenDB(), "eastmoney", "indices", fn)
|
||||
self.assertEqual(str(ctx.exception), "real upstream failure")
|
||||
|
||||
def test_classifier_exception_does_not_break_the_call(self):
|
||||
sentinel = {"ok": True}
|
||||
|
||||
def bad_classify(_result):
|
||||
raise KeyError("classifier bug")
|
||||
|
||||
result = observability.observe(self.db, "eastmoney", "indices", lambda: sentinel, classify=bad_classify)
|
||||
self.assertIs(result, sentinel)
|
||||
row = self.db.fetchone("SELECT * FROM provider_call_log")
|
||||
# A classifier bug must degrade to "ok", never silently drop the row
|
||||
# nor claim the call failed when it did not.
|
||||
self.assertEqual(row["status"], "ok")
|
||||
|
||||
def test_none_db_is_transparent_passthrough(self):
|
||||
sentinel = object()
|
||||
result = observability.observe(None, "eastmoney", "indices", lambda: sentinel)
|
||||
self.assertIs(result, sentinel)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,82 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from datahub.pipeline import RetryError
|
||||
from tests.fixtures import TRADE_DATE
|
||||
from tests.test_pipeline import make_pipeline
|
||||
|
||||
|
||||
class PipelineObservabilityTests(unittest.TestCase):
|
||||
"""HEL-543: Tushare calls must keep writing the existing `src_calls`
|
||||
record unchanged, while also feeding the new cross-provider
|
||||
`provider_call_log` / `provider_health` side channel."""
|
||||
|
||||
def test_successful_fetch_logs_to_both_src_calls_and_provider_call_log(self) -> None:
|
||||
pipe, db = make_pipeline()
|
||||
pipe.ingest_reference(TRADE_DATE)
|
||||
result = pipe.run_dataset("daily", TRADE_DATE)
|
||||
self.assertEqual(result["state"], "published")
|
||||
|
||||
src_calls = db.fetchall("SELECT * FROM src_calls WHERE provider = 'tushare' AND endpoint = 'daily'")
|
||||
self.assertTrue(any(row["ok"] == 1 for row in src_calls))
|
||||
|
||||
log = db.fetchall(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'tushare' AND interface = 'daily'"
|
||||
)
|
||||
self.assertTrue(len(log) >= 1)
|
||||
self.assertEqual(log[-1]["status"], "ok")
|
||||
|
||||
health = db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = 'tushare' AND interface = 'daily'"
|
||||
)
|
||||
self.assertIsNotNone(health)
|
||||
self.assertEqual(health["state"], "ok")
|
||||
self.assertEqual(health["consec_failures"], 0)
|
||||
|
||||
def test_failed_fetch_logs_error_to_both_channels_and_still_raises(self) -> None:
|
||||
pipe, db = make_pipeline()
|
||||
pipe.ingest_reference(TRADE_DATE)
|
||||
|
||||
def boom(dataset, params):
|
||||
raise RuntimeError("tushare upstream 500")
|
||||
|
||||
pipe.adapter.fetch = boom # type: ignore[assignment]
|
||||
|
||||
with self.assertRaises(RetryError):
|
||||
pipe.run_dataset("daily", TRADE_DATE, attempts=1)
|
||||
|
||||
src_calls = db.fetchall("SELECT * FROM src_calls WHERE provider = 'tushare' AND endpoint = 'daily' AND ok = 0")
|
||||
self.assertTrue(len(src_calls) >= 1)
|
||||
self.assertIn("tushare upstream 500", src_calls[-1]["error"])
|
||||
|
||||
log = db.fetchall(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'tushare' AND interface = 'daily' AND status != 'ok'"
|
||||
)
|
||||
self.assertTrue(len(log) >= 1)
|
||||
self.assertIn("tushare upstream 500", log[-1]["error"])
|
||||
|
||||
health = db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = 'tushare' AND interface = 'daily'"
|
||||
)
|
||||
self.assertIsNotNone(health)
|
||||
self.assertNotEqual(health["state"], "ok")
|
||||
self.assertGreaterEqual(health["consec_failures"], 1)
|
||||
|
||||
def test_provider_call_log_is_purged_by_existing_cleanup_job(self) -> None:
|
||||
pipe, db = make_pipeline()
|
||||
pipe.ingest_reference(TRADE_DATE)
|
||||
pipe.run_dataset("daily", TRADE_DATE)
|
||||
self.assertTrue(db.fetchall("SELECT * FROM provider_call_log"))
|
||||
|
||||
# Force everything to look ancient so cleanup() sweeps it.
|
||||
db.execute("UPDATE provider_call_log SET created_at = '2000-01-01T00:00:00+08:00'")
|
||||
db.execute("UPDATE src_calls SET created_at = '2000-01-01T00:00:00+08:00'")
|
||||
db.execute("UPDATE job_runs SET started_at = '2000-01-01T00:00:00+08:00'")
|
||||
|
||||
pipe.cleanup()
|
||||
self.assertEqual(db.fetchall("SELECT * FROM provider_call_log"), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,163 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from datahub.adapters.base import AdapterError
|
||||
from datahub.db import HubDB
|
||||
from datahub.realtime_serve import fetch_index_quotes, fetch_intraday, fetch_quotes
|
||||
|
||||
|
||||
class _WriteBreaksDB:
|
||||
"""Wraps a real HubDB but breaks only the write path, to prove the
|
||||
real serving path (reads/caches) is untouched by an observability
|
||||
failure while still exercising real fetch/cache code around it."""
|
||||
|
||||
def __init__(self, real: HubDB) -> None:
|
||||
self._real = real
|
||||
|
||||
def fetchall(self, sql, params=()):
|
||||
return self._real.fetchall(sql, params)
|
||||
|
||||
def fetchone(self, sql, params=()):
|
||||
return self._real.fetchone(sql, params)
|
||||
|
||||
def execute(self, sql, params=()):
|
||||
return self._real.execute(sql, params)
|
||||
|
||||
def executemany(self, sql, rows):
|
||||
return self._real.executemany(sql, rows)
|
||||
|
||||
@contextmanager
|
||||
def write(self):
|
||||
raise RuntimeError("db is not writable right now")
|
||||
yield None # pragma: no cover
|
||||
|
||||
|
||||
class RealtimeObservabilityTests(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 test_eastmoney_success_is_logged_without_changing_payload(self) -> None:
|
||||
rows = [
|
||||
{"ts_code": "000001.SH", "code": "000001", "name": "上证指数", "price": 3000.0,
|
||||
"previous_close": 2990.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
|
||||
{"ts_code": "399001.SZ", "code": "399001", "name": "深证成指", "price": 9000.0,
|
||||
"previous_close": 8990.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
|
||||
{"ts_code": "399006.SZ", "code": "399006", "name": "创业板指", "price": 1800.0,
|
||||
"previous_close": 1790.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
|
||||
]
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
|
||||
mocked.return_value.fetch_indices.return_value = rows
|
||||
payload = fetch_index_quotes(self.db)
|
||||
self.assertEqual(payload["data"], rows)
|
||||
self.assertEqual(payload["meta"]["source"], "eastmoney:ulist")
|
||||
log = self.db.fetchall("SELECT * FROM provider_call_log WHERE provider = 'eastmoney'")
|
||||
self.assertEqual(len(log), 1)
|
||||
self.assertEqual(log[0]["interface"], "indices")
|
||||
self.assertEqual(log[0]["status"], "ok")
|
||||
health = self.db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = 'eastmoney' AND interface = 'indices'"
|
||||
)
|
||||
self.assertEqual(health["state"], "ok")
|
||||
|
||||
def test_eastmoney_failure_falls_back_to_tencent_and_logs_both(self) -> None:
|
||||
tencent_rows = [
|
||||
{"ts_code": "000001.SH", "code": "000001", "name": "上证指数", "price": 3000.0,
|
||||
"previous_close": 2990.0, "quote_time_epoch": 0, "source": "tencent_qt"},
|
||||
{"ts_code": "399001.SZ", "code": "399001", "name": "深证成指", "price": 9000.0,
|
||||
"previous_close": 8990.0, "quote_time_epoch": 0, "source": "tencent_qt"},
|
||||
{"ts_code": "399006.SZ", "code": "399006", "name": "创业板指", "price": 1800.0,
|
||||
"previous_close": 1790.0, "quote_time_epoch": 0, "source": "tencent_qt"},
|
||||
]
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
|
||||
"datahub.realtime_serve.TencentAdapter"
|
||||
) as tencent:
|
||||
eastmoney.return_value.fetch_indices.side_effect = AdapterError("Eastmoney returned 0/3 indices")
|
||||
tencent.return_value.fetch_indices.return_value = tencent_rows
|
||||
payload = fetch_index_quotes(self.db)
|
||||
self.assertEqual(payload["meta"]["source"], "tencent:qt")
|
||||
self.assertEqual(payload["data"], tencent_rows)
|
||||
east_log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'eastmoney'")
|
||||
self.assertEqual(east_log["status"], "empty")
|
||||
tencent_log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'tencent'")
|
||||
self.assertEqual(tencent_log["status"], "ok")
|
||||
|
||||
def test_observability_db_failure_never_breaks_a_real_successful_fetch(self) -> None:
|
||||
rows = [
|
||||
{"ts_code": "000001.SH", "price": 3000.0, "previous_close": 2990.0, "quote_time_epoch": 0},
|
||||
{"ts_code": "399001.SZ", "price": 9000.0, "previous_close": 8990.0, "quote_time_epoch": 0},
|
||||
{"ts_code": "399006.SZ", "price": 1800.0, "previous_close": 1790.0, "quote_time_epoch": 0},
|
||||
]
|
||||
broken = _WriteBreaksDB(self.db)
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
|
||||
mocked.return_value.fetch_indices.return_value = rows
|
||||
payload = fetch_index_quotes(broken)
|
||||
self.assertEqual(payload["data"], rows)
|
||||
self.assertEqual(payload["meta"]["source"], "eastmoney:ulist")
|
||||
|
||||
def test_observability_db_failure_never_masks_a_real_source_outage(self) -> None:
|
||||
broken = _WriteBreaksDB(self.db)
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
|
||||
"datahub.realtime_serve.TencentAdapter"
|
||||
) as tencent:
|
||||
eastmoney.return_value.fetch_indices.side_effect = AdapterError("down")
|
||||
tencent.return_value.fetch_indices.side_effect = AdapterError("also down")
|
||||
with self.assertRaises(Exception):
|
||||
fetch_index_quotes(broken)
|
||||
|
||||
def test_named_quotes_records_both_providers_on_partial_merge(self) -> None:
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
|
||||
"datahub.realtime_serve.TencentAdapter"
|
||||
) as tencent:
|
||||
eastmoney.return_value.fetch_quotes.return_value = [
|
||||
{"ts_code": "000001.SZ", "close": 10, "pre_close": 9, "quote_date": "20260907"},
|
||||
]
|
||||
tencent.return_value.fetch_quotes.return_value = [
|
||||
{"ts_code": "000002.SZ", "close": 20, "pre_close": 19, "quote_date": "20260907"},
|
||||
]
|
||||
payload = fetch_quotes(self.db, ["000001.SZ", "000002.SZ"])
|
||||
self.assertEqual(payload["meta"]["complete"], True)
|
||||
east_log = self.db.fetchone(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'named_quotes'"
|
||||
)
|
||||
self.assertEqual(east_log["status"], "ok")
|
||||
tencent_log = self.db.fetchone(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'tencent' AND interface = 'named_quotes'"
|
||||
)
|
||||
self.assertEqual(tencent_log["status"], "ok")
|
||||
|
||||
def test_intraday_success_is_logged_as_ok(self) -> None:
|
||||
payload_data = {
|
||||
"entity_type": "stock", "ts_code": "601318.SH", "trade_date": "2026-09-07",
|
||||
"previous_close": 55.8, "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9}],
|
||||
}
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
|
||||
mocked.return_value.fetch_intraday.return_value = payload_data
|
||||
payload = fetch_intraday(self.db, "601318.SH")
|
||||
self.assertEqual(payload["data"], payload_data)
|
||||
log = self.db.fetchone(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'intraday'"
|
||||
)
|
||||
self.assertEqual(log["status"], "ok")
|
||||
|
||||
def test_intraday_failure_is_logged_as_empty(self) -> None:
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
|
||||
mocked.return_value.fetch_intraday.side_effect = AdapterError("No intraday chart data returned")
|
||||
with self.assertRaises(Exception):
|
||||
fetch_intraday(self.db, "000001.SZ")
|
||||
log = self.db.fetchone(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'intraday'"
|
||||
)
|
||||
self.assertEqual(log["status"], "empty")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user