From 8f5876db2c3cde85900142911d514817cb8ee547 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Apr 2026 09:35:52 +0800 Subject: [PATCH] fix: surface clearer longbridge token expiry errors --- src/quant_platform_kit/longbridge/auth.py | 49 ++++++++++++++---- tests/test_longbridge_auth.py | 63 +++++++++++++++++++++++ 2 files changed, 102 insertions(+), 10 deletions(-) diff --git a/src/quant_platform_kit/longbridge/auth.py b/src/quant_platform_kit/longbridge/auth.py index bd7b022..1a7c427 100644 --- a/src/quant_platform_kit/longbridge/auth.py +++ b/src/quant_platform_kit/longbridge/auth.py @@ -5,6 +5,7 @@ import hmac import json import time +from datetime import datetime, timezone from typing import Any, Callable @@ -42,6 +43,26 @@ def _longport_sign(method: str, uri: str, headers: dict[str, str], params: str, return f"HMAC-SHA256 SignedHeaders=authorization;x-api-key;x-timestamp, Signature={signature}" +def _decode_token_expiry(token: str) -> float | None: + try: + parts = token.split(".") + if len(parts) <= 1: + return None + payload_b64 = parts[1] + padded_payload = payload_b64 + "=" * (-len(payload_b64) % 4) + payload = json.loads(base64.urlsafe_b64decode(padded_payload).decode("utf-8")) + expiry = payload.get("exp") + if expiry is None: + return None + return float(expiry) + except Exception: + return None + + +def _format_expiry(expiry_timestamp: float) -> str: + return datetime.fromtimestamp(expiry_timestamp, timezone.utc).isoformat() + + def refresh_token_if_needed( current_token: str, *, @@ -53,19 +74,20 @@ def refresh_token_if_needed( requests_module: Any | None = None, secret_client_factory: Callable[[], Any] | None = None, ) -> str: + expiry_timestamp = _decode_token_expiry(current_token) + now = time.time() + if not app_key or not app_secret: + if expiry_timestamp is not None and expiry_timestamp <= now: + raise RuntimeError( + "LongPort token in secret " + f"'{secret_name}' expired at {_format_expiry(expiry_timestamp)} " + "and cannot be refreshed because LONGPORT_APP_KEY/LONGPORT_APP_SECRET is missing." + ) return current_token - try: - parts = current_token.split(".") - if len(parts) > 1: - payload_b64 = parts[1] - padded_payload = payload_b64 + "=" * (-len(payload_b64) % 4) - payload = json.loads(base64.urlsafe_b64decode(padded_payload).decode("utf-8")) - if (payload.get("exp", 0) - time.time()) / 86400 > refresh_threshold_days: - return current_token - except Exception: - pass + if expiry_timestamp is not None and (expiry_timestamp - now) / 86400 > refresh_threshold_days: + return current_token if requests_module is None: import requests as requests_module @@ -83,6 +105,13 @@ def refresh_token_if_needed( timeout=15, ).json() if response.get("code") != 0: + if expiry_timestamp is not None and expiry_timestamp <= now: + code = response.get("code") + message = response.get("message") or "unknown error" + raise RuntimeError( + f"LongPort token in secret '{secret_name}' expired at {_format_expiry(expiry_timestamp)}; " + f"refresh failed with code {code}: {message}" + ) return current_token new_token = response["data"]["token"] diff --git a/tests/test_longbridge_auth.py b/tests/test_longbridge_auth.py index 21b38d2..06528b0 100644 --- a/tests/test_longbridge_auth.py +++ b/tests/test_longbridge_auth.py @@ -3,6 +3,7 @@ import base64 import json import sys +import time import types import unittest from unittest.mock import patch @@ -62,6 +63,17 @@ def json(): return Response() +class FakeFailedRequests: + @staticmethod + def get(url, headers, timeout): + class Response: + @staticmethod + def json(): + return {"code": 401003, "message": "token expired", "data": None} + + return Response() + + class LongBridgeAuthTests(unittest.TestCase): def test_fetch_token_from_secret_reads_latest_version(self) -> None: client = FakeSecretClient("token-abc") @@ -109,6 +121,57 @@ def test_refresh_token_if_needed_persists_new_token(self) -> None: self.assertEqual(client.created_parent, "projects/demo/secrets/token") self.assertEqual(client.destroyed, ["projects/demo/secrets/token/versions/1"]) + def test_refresh_token_if_needed_raises_clear_error_when_expired_and_refresh_fails(self) -> None: + payload = {"exp": 1} + encoded = base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8").rstrip("=") + token = f"aaa.{encoded}.bbb" + + with self.assertRaises(RuntimeError) as context: + refresh_token_if_needed( + token, + project_id="demo", + secret_name="longport_token_sg", + app_key="key", + app_secret="secret", + requests_module=FakeFailedRequests, + ) + + self.assertIn("longport_token_sg", str(context.exception)) + self.assertIn("refresh failed with code 401003", str(context.exception)) + + def test_refresh_token_if_needed_returns_same_token_when_refresh_fails_but_token_not_expired(self) -> None: + payload = {"exp": int(time.time()) + 86400} + encoded = base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8").rstrip("=") + token = f"aaa.{encoded}.bbb" + + refreshed = refresh_token_if_needed( + token, + project_id="demo", + secret_name="token", + app_key="key", + app_secret="secret", + refresh_threshold_days=30, + requests_module=FakeFailedRequests, + ) + + self.assertEqual(refreshed, token) + + def test_refresh_token_if_needed_raises_clear_error_when_expired_and_app_credentials_missing(self) -> None: + payload = {"exp": 1} + encoded = base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8").rstrip("=") + token = f"aaa.{encoded}.bbb" + + with self.assertRaises(RuntimeError) as context: + refresh_token_if_needed( + token, + project_id="demo", + secret_name="longport_token_sg", + app_key="", + app_secret="", + ) + + self.assertIn("LONGPORT_APP_KEY/LONGPORT_APP_SECRET is missing", str(context.exception)) + def test_build_contexts_uses_longport_openapi(self) -> None: longport_module = types.ModuleType("longport") openapi_module = types.ModuleType("longport.openapi")