dekun
97 lines
3.8 KiB
Python
97 lines
3.8 KiB
Python
"""hub_system_logs_lib 单元测试."""
|
|
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,
|
|
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"])
|
|
keys = [t["key"] for t in meta["targets"]]
|
|
self.assertEqual(keys, ["binance", "gate", "okx", "hub"])
|
|
|
|
def test_tail_lines_reads_last_lines(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "demo-out.log"
|
|
path.write_text("\n".join(f"line-{i}" for i in range(1, 11)), encoding="utf-8")
|
|
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_with_resolved_paths(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
logs_dir = Path(tmp)
|
|
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"])
|