94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
# Test aider_event_service | 2026-04-20 @ Asia/Taipei
|
||
import pytest
|
||
from datetime import datetime, timezone, timedelta
|
||
from src.services.aider_event_service import (
|
||
classify_severity, should_create_incident, build_signal_data,
|
||
)
|
||
from src.models.aider import AiderEventIn
|
||
|
||
TAIPEI = timezone(timedelta(hours=8))
|
||
|
||
|
||
def _ev(t, p=None, sid="s1"):
|
||
return AiderEventIn(
|
||
ts=datetime.now(TAIPEI), session_id=sid,
|
||
host="ogt-mac", type=t, payload=p or {},
|
||
)
|
||
|
||
|
||
def test_classify_error_is_warning():
|
||
# aider "error" event → signal severity "warning" → Severity.P2
|
||
assert classify_severity(_ev("error")) == "warning"
|
||
|
||
|
||
def test_classify_silent_timeout_is_info():
|
||
# silent_timeout → "info" → P3
|
||
assert classify_severity(_ev("silent_timeout")) == "info"
|
||
|
||
|
||
def test_classify_session_end_nonzero_exit_is_warning():
|
||
assert classify_severity(_ev("session_end", {"exit_code": 1})) == "warning"
|
||
|
||
|
||
def test_classify_session_end_multi_error_is_high():
|
||
# 3+ errors + nonzero exit → "high" → P1
|
||
assert classify_severity(_ev("session_end", {"exit_code": 1, "error_count": 5})) == "high"
|
||
|
||
|
||
def test_classify_session_end_clean_is_none():
|
||
ev = _ev("session_end", {"exit_code": 0, "error_count": 0})
|
||
assert classify_severity(ev) is None
|
||
assert should_create_incident(ev) is False
|
||
|
||
|
||
def test_classify_session_start_is_none():
|
||
assert should_create_incident(_ev("session_start")) is False
|
||
|
||
|
||
def test_classify_error_triggers_incident():
|
||
assert should_create_incident(_ev("error")) is True
|
||
|
||
|
||
def test_build_signal_data_fingerprint_per_session_type():
|
||
# 同 session + 同 type → 同 fingerprint（讓既有 3min debounce 生效）
|
||
sd1 = build_signal_data(_ev("error", {"cwd": "/a", "model": "m"}, sid="sx"))
|
||
sd2 = build_signal_data(_ev("error", {"cwd": "/a", "model": "m"}, sid="sx"))
|
||
assert sd1["fingerprint"] == sd2["fingerprint"]
|
||
assert sd1["fingerprint"].startswith("aider:sx:error")
|
||
|
||
|
||
def test_build_signal_data_different_sessions_different_fp():
|
||
sd_a = build_signal_data(_ev("error", sid="sA"))
|
||
sd_b = build_signal_data(_ev("error", sid="sB"))
|
||
assert sd_a["fingerprint"] != sd_b["fingerprint"]
|
||
|
||
|
||
def test_build_signal_data_shape():
|
||
ev = _ev("error", {"cwd": "/Users/ogt/awoooi", "model": "elephant-alpha",
|
||
"message": "api timeout", "kind": "api_rate_limit"})
|
||
sd = build_signal_data(ev)
|
||
# 必要 keys
|
||
for k in ("alert_name", "severity", "source", "fingerprint", "target",
|
||
"labels", "annotations"):
|
||
assert k in sd, f"missing {k}"
|
||
assert sd["source"] == "manual"
|
||
assert sd["target"] == "awoooi" # 從 cwd basename
|
||
assert sd["labels"]["session_id"] == "s1"
|
||
assert sd["labels"]["model"] == "elephant-alpha"
|
||
|
||
|
||
def test_build_signal_data_redacts_secrets_in_annotations():
|
||
"""secret 不可進 annotations/labels"""
|
||
ev = _ev("error", {"cwd": "/r", "model": "m",
|
||
"message": "key sk-or-v1-abcdef0123456789ABCDEFghijklmnopqrstuv fail"})
|
||
sd = build_signal_data(ev)
|
||
# annotations 可能含 message — 需已遮罩
|
||
annot_str = str(sd["annotations"])
|
||
assert "sk-or-v1-abcdef" not in annot_str
|
||
assert "<redacted:" in annot_str
|
||
|
||
|
||
def test_build_signal_data_returns_none_for_non_incident_types():
|
||
# session_start 不該建 signal
|
||
assert build_signal_data(_ev("session_start")) is None
|