49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from backend.data import DataPolicyError, DataSourcePolicy, build_data_gateway
|
|
|
|
|
|
class DataGatewayTests(unittest.TestCase):
|
|
def test_policy_allows_registered_calculation_source(self) -> None:
|
|
policy = DataSourcePolicy.load()
|
|
contract = policy.assert_allowed(
|
|
"market.stock_daily", "tushare", "calculation"
|
|
)
|
|
self.assertEqual(contract.primary, "tushare")
|
|
|
|
def test_policy_rejects_public_web_source_for_calculation(self) -> None:
|
|
policy = DataSourcePolicy.load()
|
|
with self.assertRaises(DataPolicyError):
|
|
policy.assert_allowed(
|
|
"observation.realtime_indices", "eastmoney", "calculation"
|
|
)
|
|
|
|
def test_policy_rejects_blocked_dataset(self) -> None:
|
|
policy = DataSourcePolicy.load()
|
|
with self.assertRaises(DataPolicyError):
|
|
policy.assert_allowed("market.level2", "unresolved", "display")
|
|
|
|
def test_gateway_uses_live_token_supplier_and_shared_ifind(self) -> None:
|
|
token = {"value": "first"}
|
|
gateway = build_data_gateway(
|
|
{"ifind_refresh_token": "refresh", "ifind_access_token": "access"},
|
|
lambda: token["value"],
|
|
)
|
|
self.assertEqual(gateway.tushare().token, "first")
|
|
token["value"] = "second"
|
|
self.assertEqual(gateway.tushare().token, "second")
|
|
self.assertIs(gateway.chart_data.ifind, gateway.ifind)
|
|
|
|
def test_server_has_no_direct_runtime_tushare_construction(self) -> None:
|
|
from pathlib import Path
|
|
|
|
source = (Path(__file__).resolve().parents[1] / "server.py").read_text(encoding="utf-8")
|
|
self.assertEqual(source.count("TushareClient(self.token)"), 1)
|
|
self.assertIn("return gateway.tushare()", source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|