dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix hub system logs by resolving PM2 log paths from jlist and glob fallback.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-09 10:38:24 +08:00
parent 5214794f95
commit d092f213a7
2 changed files with 150 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_hub_system_logs_lib.py
+62 -13
View File
@@ -4,16 +4,22 @@ from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from lib.hub import hub_system_logs_lib as logs_lib
from lib.hub.hub_system_logs_lib import (
load_system_logs,
log_file_paths,
resolve_log_paths,
system_logs_meta,
tail_lines,
)
class HubSystemLogsLibTest(unittest.TestCase):
def setUp(self):
logs_lib._path_cache.clear()
logs_lib._path_cache_at = 0.0
def test_system_logs_meta(self):
meta = system_logs_meta()
self.assertTrue(meta["ok"])
@@ -27,21 +33,64 @@ class HubSystemLogsLibTest(unittest.TestCase):
out = tail_lines(path, lines=3)
self.assertEqual(out.splitlines(), ["line-8", "line-9", "line-10"])
def test_resolve_log_paths_from_pm2_jlist(self):
with tempfile.TemporaryDirectory() as tmp:
logs_dir = Path(tmp)
out_file = logs_dir / "crypto-binance-out-0.log"
err_file = logs_dir / "crypto-binance-error-0.log"
out_file.write_text("stdout line", encoding="utf-8")
err_file.write_text("stderr line", encoding="utf-8")
payload = [
{
"name": "crypto_binance",
"pm2_env": {
"pm_out_log_path": str(out_file),
"pm_err_log_path": str(err_file),
},
}
]
with patch.object(logs_lib, "_pm2_jlist", return_value=payload):
out_path, err_path = resolve_log_paths("crypto_binance")
self.assertEqual(out_path, out_file)
self.assertEqual(err_path, err_file)
def test_resolve_log_paths_glob_fallback(self):
with tempfile.TemporaryDirectory() as tmp:
logs_dir = Path(tmp)
out_file = logs_dir / "crypto-gate-out-1.log"
err_file = logs_dir / "crypto-gate-error-1.log"
out_file.write_text("gate out", encoding="utf-8")
err_file.write_text("gate err", encoding="utf-8")
with patch.object(logs_lib, "pm2_logs_dir", return_value=logs_dir):
with patch.object(logs_lib, "_pm2_jlist", return_value=[]):
out_path, err_path = resolve_log_paths("crypto_gate")
self.assertEqual(out_path, out_file)
self.assertEqual(err_path, err_file)
def test_load_system_logs_unknown(self):
with self.assertRaises(KeyError):
load_system_logs("unknown")
def test_load_system_logs_missing_files(self):
def test_load_system_logs_with_resolved_paths(self):
with tempfile.TemporaryDirectory() as tmp:
logs_dir = Path(tmp)
out_path, err_path = log_file_paths("crypto_gate")
# monkeypatch by writing into expected structure under temp is hard;
# just verify shape with real call (files may be absent on dev machine).
payload = load_system_logs("gate", lines=50)
self.assertTrue(payload["ok"])
self.assertEqual(payload["key"], "gate")
self.assertIn("out", payload)
self.assertIn("err", payload)
self.assertIsInstance(payload["out"], str)
self.assertIsInstance(payload["err"], str)
_ = out_path, err_path
out_file = logs_dir / "manual-trading-hub-out-6.log"
err_file = logs_dir / "manual-trading-hub-error-6.log"
out_file.write_text("hub stdout", encoding="utf-8")
err_file.write_text("hub stderr", encoding="utf-8")
payload = [
{
"name": "manual-trading-hub",
"pm2_env": {
"pm_out_log_path": str(out_file),
"pm_err_log_path": str(err_file),
},
}
]
with patch.object(logs_lib, "_pm2_jlist", return_value=payload):
data = load_system_logs("hub", lines=50)
self.assertTrue(data["ok"])
self.assertIn("hub stdout", data["out"])
self.assertIn("hub stderr", data["err"])
self.assertTrue(data["out_exists"])
self.assertTrue(data["err_exists"])