Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -334,6 +334,18 @@ def _runtime_error_notification_message(exc: Exception, *, route_label: str) ->
error_text = f"{type(exc).__name__}: {exc}"
if len(error_text) > 1200:
error_text = error_text[:1197] + "..."
if str(NOTIFY_LANG or "").strip().lower().startswith("zh"):
return "\n".join(
(
"LongBridge 策略运行失败",
f"服务: {os.getenv('K_SERVICE') or SECRET_NAME or 'longbridge-platform'}",
f"版本: {os.getenv('K_REVISION') or '<unknown>'}",
f"路由: {route_label}",
f"策略: {STRATEGY_PROFILE}",
f"账户范围: {ACCOUNT_REGION}",
f"错误: {error_text}",
)
)
return "\n".join(
(
"LongBridge strategy run failed",
Expand Down
34 changes: 30 additions & 4 deletions tests/test_request_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@


@contextmanager
def install_stub_modules():
def install_stub_modules(*, notify_lang="en"):
flask_module = types.ModuleType("flask")

class Flask:
Expand Down Expand Up @@ -67,7 +67,7 @@ def run(self, *args, **kwargs):
market_timezone="Asia/Hong_Kong",
symbol_suffix=".HK",
trading_currency="HKD",
notify_lang="en",
notify_lang=notify_lang,
tg_token=None,
tg_chat_id="shared-chat-id",
dry_run_only=False,
Expand Down Expand Up @@ -186,8 +186,8 @@ def run(self, *args, **kwargs):
sys.modules[name] = previous


def load_module():
with install_stub_modules():
def load_module(*, notify_lang="en"):
with install_stub_modules(notify_lang=notify_lang):
with patch.dict(
os.environ,
{
Expand Down Expand Up @@ -277,6 +277,32 @@ def fake_post(_url, *, json, timeout):
self.assertIn("LongBridge strategy run failed", observed["payloads"][0][0]["text"])
self.assertIn("RuntimeError: boom", observed["payloads"][0][0]["text"])

def test_handle_trigger_runtime_error_fallback_uses_chinese_copy(self):
module = load_module(notify_lang="zh")
observed = {"payloads": []}

class FakeResponse:
status_code = 200

def fake_post(_url, *, json, timeout):
observed["payloads"].append((json, timeout))
return FakeResponse()

module.TG_TOKEN = "token-1"
module.TG_CHAT_ID = "chat-1"
module.requests.post = fake_post
module.run_strategy = lambda: (_ for _ in ()).throw(RuntimeError("boom"))

with module.app.test_request_context("/", method="POST"):
body, status = module.handle_trigger()

self.assertEqual(status, 500)
self.assertEqual(body, "Error")
text = observed["payloads"][0][0]["text"]
self.assertIn("LongBridge 策略运行失败", text)
self.assertIn("服务:", text)
self.assertIn("错误: RuntimeError: boom", text)

def test_handle_trigger_allows_get(self):
module = load_module()
observed = {"called": False}
Expand Down