from __future__ import annotations import tempfile import unittest from pathlib import Path from unittest.mock import patch from datahub.adapters.base import AdapterError from datahub.adapters.eastmoney import HIS_TRENDS_URL, TRENDS_URL, EastmoneyAdapter from datahub.db import HubDB from datahub.realtime_serve import fetch_intraday from datahub.serving import ApiError, V1API from datahub.timeutil import now_shanghai, yyyymmdd class FakeEastmoney(EastmoneyAdapter): def __init__(self) -> None: super().__init__(timeout=2) self.urls: list[str] = [] def _get_json(self, url, params, referer): self.urls.append(f"{url}|{params.get('ndays')}") if url == TRENDS_URL: return {"data": {"name": "中国平安", "code": "601318", "preClose": 56.36, "trends": []}} if url == HIS_TRENDS_URL: return { "data": { "name": "中国平安", "code": "601318", "preClose": 55.8, "trends": [ "2026-09-07 09:30,55.80,55.90,56.00,55.70,100,5580.00,55.900", "2026-09-07 15:00,56.10,56.20,56.30,56.00,200,11240.00,56.150", "2026-09-08 09:30,0,0,0,0,0,0.00,0", ], } } raise AdapterError(f"unexpected url {url}") class EastmoneyIntradayLookbackTests(unittest.TestCase): def test_empty_today_uses_latest_available_session(self): adapter = FakeEastmoney() payload = adapter.fetch_intraday("601318.SH") self.assertEqual(adapter.urls, [f"{TRENDS_URL}|1", f"{TRENDS_URL}|5", f"{HIS_TRENDS_URL}|5"]) self.assertEqual(payload["trade_date"], "2026-09-07") self.assertEqual([point["time"] for point in payload["points"]], ["09:30", "15:00"]) self.assertEqual(payload["points"][0]["close"], 55.9) def test_preferred_date_keeps_that_session(self): adapter = FakeEastmoney() payload = adapter.fetch_intraday("601318.SH", "20260907") self.assertEqual(payload["trade_date"], "2026-09-07") self.assertEqual(len(payload["points"]), 2) class IntradayLkgTests(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_source_failure_returns_last_known_good(self): from datahub.realtime_serve import _envelope, _write_cache payload = _envelope( { "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}], }, { "tier": "provisional", "trade_date": "20260907", "source": "eastmoney:trends2", "stale": False, }, ) _write_cache(self.db, "intraday:601318.SH:today", payload, 20, "eastmoney:trends2") self.db.execute( "UPDATE rt_cache SET expires_at = ? WHERE cache_key = ?", ("2000-01-01T00:00:00+08:00", "intraday:601318.SH:today"), ) with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: mocked.return_value.fetch_intraday.side_effect = AdapterError("down") recovered = fetch_intraday(self.db, "601318.SH") self.assertTrue(recovered["meta"]["stale"]) self.assertEqual(recovered["data"]["points"][0]["close"], 55.9) def test_source_failure_without_lkg_raises(self): with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: mocked.return_value.fetch_intraday.side_effect = AdapterError("down") with self.assertRaises(Exception) as ctx: fetch_intraday(self.db, "000001.SZ") self.assertIn("intraday unavailable", str(ctx.exception)) class ServingIntradayDateTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() self.db = HubDB(Path(self.tmp.name) / "hub.db") self.api = V1API(self.db, pipeline=None, settings=None) def tearDown(self) -> None: self.tmp.cleanup() def _assert_usable_intraday(self, payload: dict) -> None: data = payload["data"] points = [point for point in data.get("points") or [] if float(point.get("close") or 0) > 0] self.assertGreaterEqual(len(points), 1) self.assertTrue(str(data.get("trade_date") or "")) self.assertFalse((payload.get("meta") or {}).get("stale")) def test_serving_omitted_or_empty_date_uses_today_and_returns_points(self) -> None: today = yyyymmdd(now_shanghai()) omitted = self.api.handle("/v1/intraday/points", {"code": ["601318"]}) empty = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": [""]}) explicit = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": [today]}) self._assert_usable_intraday(omitted) self._assert_usable_intraday(empty) self._assert_usable_intraday(explicit) self.assertEqual(omitted["data"]["trade_date"], empty["data"]["trade_date"]) self.assertEqual(explicit["data"]["trade_date"], omitted["data"]["trade_date"]) def test_serving_normalizes_empty_date_to_today_and_keeps_history(self) -> None: today = yyyymmdd(now_shanghai()) captured: list[str] = [] def fake_fetch(db, code, date=""): captured.append(date) return { "schema_version": 1, "data": { "trade_date": f"{date[:4]}-{date[4:6]}-{date[6:8]}", "points": [{"date": f"{date[:4]}-{date[4:6]}-{date[6:8]}", "time": "09:30", "close": 55.9}], }, "meta": {"stale": False, "trade_date": date}, } with patch("datahub.realtime_serve.fetch_intraday", side_effect=fake_fetch): omitted = self.api.handle("/v1/intraday/points", {"code": ["601318"]}) empty = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": [" "]}) history = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": ["20260907"]}) self.assertEqual(captured, [today, today, "20260907"]) self.assertEqual(omitted["data"]["trade_date"], f"{today[:4]}-{today[4:6]}-{today[6:8]}") self.assertEqual(empty["data"]["trade_date"], omitted["data"]["trade_date"]) self.assertEqual(history["data"]["trade_date"], "2026-09-07") def test_serving_invalid_date_is_invalid_argument(self) -> None: with self.assertRaises(ApiError) as ctx: self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": ["not-a-date"]}) self.assertEqual(ctx.exception.code, "INVALID_ARGUMENT") self.assertIn("invalid trade_date", ctx.exception.message) def test_serving_missing_code_is_invalid_argument(self) -> None: with self.assertRaises(ApiError) as ctx: self.api.handle("/v1/intraday/points", {"date": [yyyymmdd(now_shanghai())]}) self.assertEqual(ctx.exception.code, "INVALID_ARGUMENT") self.assertIn("code is required", ctx.exception.message) def test_serving_no_data_keeps_source_unavailable(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(ApiError) as ctx: self.api.handle("/v1/intraday/points", {"code": ["000001"]}) self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE") self.assertIn("intraday unavailable", ctx.exception.message) if __name__ == "__main__": unittest.main()