142 lines
5.6 KiB
Python
142 lines
5.6 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
import core.toolfail as tf # noqa: E402
|
|
|
|
NOW = datetime.now(timezone.utc)
|
|
|
|
|
|
class _FakeSession:
|
|
"""按 SQL 关键字分流三段查询:messages / tool_malformed / run_error。"""
|
|
|
|
def __init__(self, msg_rows, malformed_rows, run_rows):
|
|
self._msg_rows = msg_rows
|
|
self._malformed_rows = malformed_rows
|
|
self._run_rows = run_rows
|
|
|
|
def execute(self, clause, params=None):
|
|
sql = str(clause)
|
|
if "tool_malformed" in sql:
|
|
rows = self._malformed_rows
|
|
elif "run_error" in sql:
|
|
rows = self._run_rows
|
|
else:
|
|
rows = self._msg_rows
|
|
return SimpleNamespace(fetchall=lambda: rows)
|
|
|
|
|
|
def _scan(msg_rows, malformed_rows, run_rows=None, **kw):
|
|
@contextmanager
|
|
def fake_scope():
|
|
yield _FakeSession(msg_rows, malformed_rows, run_rows or [])
|
|
|
|
with patch.object(tf, "session_scope", fake_scope):
|
|
return tf.scan_tool_failures(**kw)
|
|
|
|
|
|
class TestToolfailMalformed(unittest.TestCase):
|
|
def test_malformed_rows_cluster(self):
|
|
"""usage_events 的 tool_malformed 行聚成 kind=malformed 的 cluster。"""
|
|
err = "Expecting value: line 1 column 1 (char 0)"
|
|
rows = [
|
|
("t1", "u1", NOW - timedelta(hours=1), "write", err, "].cells[1]", '"}'),
|
|
("t2", "u1", NOW - timedelta(hours=2), "write", err, "].merge(", '"}'),
|
|
]
|
|
out = _scan([], rows, days=7, min_count=2, min_tasks=2)
|
|
self.assertEqual(len(out), 1)
|
|
c = out[0]
|
|
self.assertEqual(c["tool"], "write")
|
|
self.assertEqual(c["kind"], "malformed")
|
|
self.assertEqual(c["count"], 2)
|
|
self.assertEqual(c["count_24h"], 2)
|
|
self.assertEqual(c["task_count"], 2)
|
|
self.assertIn("…", c["sample"]) # sample = head … tail
|
|
|
|
def test_err_signature_normalized(self):
|
|
"""报错里的行号/列号数字被归一,不同位置的同类报错聚成一条。"""
|
|
rows = [
|
|
("t1", "u1", NOW, "write", "Expecting ',' delimiter: line 1 column 57 (char 56)", "h", "t"),
|
|
("t2", "u2", NOW, "write", "Expecting ',' delimiter: line 3 column 9 (char 88)", "h", "t"),
|
|
]
|
|
out = _scan([], rows, days=7, min_count=2, min_tasks=2)
|
|
self.assertEqual(len(out), 1)
|
|
|
|
def test_merges_with_message_clusters(self):
|
|
"""两段数据源并存互不干扰:messages 报错与 malformed 各出各的 cluster。"""
|
|
msg_rows = [
|
|
("t1", "u1", NOW, "glob", "[Error] base path not found"),
|
|
("t2", "u2", NOW, "glob", "[Error] base path not found"),
|
|
]
|
|
mal_rows = [
|
|
("t3", "u1", NOW, "write", "Expecting value", "h", "t"),
|
|
("t4", "u2", NOW, "write", "Expecting value", "h", "t"),
|
|
]
|
|
out = _scan(msg_rows, mal_rows, days=7, min_count=2, min_tasks=2)
|
|
kinds = sorted(c["kind"] for c in out)
|
|
self.assertEqual(kinds, ["error", "malformed"])
|
|
|
|
def test_quiet_cluster_daily_bucket(self):
|
|
"""3 天前的畸形记录:count_24h=0(已安静),落在对应历史桶。"""
|
|
rows = [
|
|
("t1", "u1", NOW - timedelta(days=3, hours=1), "write", "Expecting value", "h", "t"),
|
|
("t2", "u2", NOW - timedelta(days=3, hours=2), "write", "Expecting value", "h", "t"),
|
|
]
|
|
out = _scan([], rows, days=7, min_count=2, min_tasks=2)
|
|
c = out[0]
|
|
self.assertEqual(c["count_24h"], 0)
|
|
self.assertEqual(sum(c["daily"]), 2)
|
|
self.assertEqual(c["daily"][-4], 2) # 尾桶=近24h,往前退 3 桶
|
|
|
|
|
|
class TestToolfailRunError(unittest.TestCase):
|
|
def test_run_error_rows_cluster(self):
|
|
"""usage_events 的 run_error 行聚成 kind=run / tool=(run) 的 cluster。"""
|
|
err = "RateLimitError: litellm.RateLimitError: ZaiException - 余额不足或无可用资源包,请充值。"
|
|
rows = [
|
|
("t1", "u1", NOW - timedelta(hours=1), err),
|
|
("t1", "u1", NOW - timedelta(minutes=30), err),
|
|
("t2", "u2", NOW - timedelta(minutes=10), err),
|
|
]
|
|
out = _scan([], [], rows, days=7, min_count=3, min_tasks=2)
|
|
self.assertEqual(len(out), 1)
|
|
c = out[0]
|
|
self.assertEqual(c["tool"], "(run)")
|
|
self.assertEqual(c["kind"], "run")
|
|
self.assertEqual(c["count"], 3)
|
|
self.assertEqual(c["task_count"], 2)
|
|
self.assertIn("余额不足", c["sample"])
|
|
|
|
|
|
class TestProviderCriticalAlert(unittest.TestCase):
|
|
def setUp(self):
|
|
tf._alerted_at.clear()
|
|
|
|
def test_pattern_hit_and_cooldown(self):
|
|
"""余额类错误命中;同签名冷却期内第二次不再发。"""
|
|
err = "RateLimitError: ZaiException - 余额不足或无可用资源包,请充值。"
|
|
sent = []
|
|
with patch("tools.send_email.smtp_configured", return_value=True), \
|
|
patch("tools.send_email.send_email_smtp",
|
|
side_effect=lambda *a, **k: sent.append(a)), \
|
|
patch.dict("os.environ", {"ZCBOT_DEVELOPER_EMAIL": "dev@x.com"}):
|
|
self.assertTrue(tf.alert_provider_critical(err, task_id="t1"))
|
|
self.assertFalse(tf.alert_provider_critical(err, task_id="t2")) # 冷却
|
|
self.assertEqual(len(sent), 1)
|
|
|
|
def test_non_critical_ignored(self):
|
|
"""普通异常不告警。"""
|
|
self.assertFalse(tf.alert_provider_critical("ValueError: bad input"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|