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>
83 lines
3.3 KiB
Python
83 lines
3.3 KiB
Python
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()
|