from __future__ import annotations

import ast
import unittest
from pathlib import Path
from unittest.mock import patch

from backend.data.datahub.bridge import DatahubAwareTushareClient, DatahubBridge
from backend.data.datahub.client import DatahubClient
from backend.data.datahub.settings import DATASETS, DatahubSettings, DatasetFlags
from backend.data.providers.tushare_transport import TushareError
from tests.test_datahub_bridge import FakeClient, FakeLegacy, flags


ROOT = Path(__file__).resolve().parents[1]
BLOCKED_HOSTS = (
    "api.tushare.pro",
    "push2.eastmoney.com",
    "push2delay.eastmoney.com",
    "push2his.eastmoney.com",
    "push2ex.eastmoney.com",
    "qt.gtimg.cn",
    "hq.sinajs.cn",
    "10jqka.com.cn",
    "xuangubao.cn",
)


class HubExclusiveWebsiteTests(unittest.TestCase):
    def test_query_never_calls_website_tushare_transport(self) -> None:
        client = FakeClient()
        legacy = FakeLegacy(TushareError("website tushare must stay dark"))
        wrapped = DatahubAwareTushareClient(
            legacy,
            DatahubBridge(flags(daily=(True, False)), client),
        )
        rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,amount")
        self.assertEqual(rows[0]["amount"], 2000.0)
        self.assertEqual(legacy.calls, [])

    def test_blocked_external_hosts_still_read_hub(self) -> None:
        settings = DatahubSettings(
            base_url="http://127.0.0.1:8766",
            token="hub-token",
            datasets={name: DatasetFlags(name, read=True) for name in DATASETS},
        )

        def blocked_urlopen(request, timeout=None):
            url = str(getattr(request, "full_url", None) or request)
            if any(host in url for host in BLOCKED_HOSTS):
                raise AssertionError(f"website opened blocked host: {url}")
            if "127.0.0.1:8766" in url or "v1/bars/daily" in url:
                class _Resp:
                    status = 200

                    def read(self):
                        return (
                            b'{"schema_version":1,"data":[{"ts_code":"600000.SH","trade_date":"20240902",'
                            b'"close":10.2,"volume":100000,"amount":2000000}],'
                            b'"meta":{"stale":false,"staleness_seconds":0,"source":"datahub"}}'
                        )

                    def __enter__(self):
                        return self

                    def __exit__(self, *args):
                        return False

                return _Resp()
            raise AssertionError(f"unexpected url: {url}")

        hub_client = DatahubClient(settings, urlopen=blocked_urlopen)
        legacy = FakeLegacy(TushareError("blocked"))
        wrapped = DatahubAwareTushareClient(legacy, DatahubBridge(settings, hub_client))
        with patch("urllib.request.urlopen", blocked_urlopen):
            rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,close,amount")
        self.assertEqual(rows[0]["close"], 10.2)
        self.assertEqual(rows[0]["amount"], 2000.0)
        self.assertEqual(legacy.calls, [])

    def test_website_runtime_does_not_call_blocked_hosts_from_gateway(self) -> None:
        gateway_src = (ROOT / "backend" / "data" / "gateway.py").read_text(encoding="utf-8")
        self.assertIn("legacy.realtime_aggregator = None", gateway_src)
        self.assertIn("DatahubAwareTushareClient", gateway_src)

    def test_bridge_query_has_no_legacy_call(self) -> None:
        source = (ROOT / "backend" / "data" / "datahub" / "bridge.py").read_text(encoding="utf-8")
        tree = ast.parse(source)
        query_fn = next(
            node
            for node in tree.body
            if isinstance(node, ast.ClassDef) and node.name == "DatahubBridge"
            for item in node.body
            if isinstance(item, ast.FunctionDef) and item.name == "query"
        )
        called = [
            ast.unparse(item.func) if hasattr(ast, "unparse") else ""
            for item in ast.walk(query_fn)
            if isinstance(item, ast.Call)
        ]
        self.assertTrue(any("query_api" in text for text in called))
        self.assertFalse(any("legacy_query" in text for text in called))


if __name__ == "__main__":
    unittest.main()