4f1c8ae473
- Fix 35 Python ruff errors (B904, F841, E722, E741, B007, B008) - Add eslint config for lewooogo-core package - Update pyproject.toml to new ruff lint config format - Relax frontend eslint rules to warnings for unused vars - Allow console.* for debugging (TODO: unified logger) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
244 lines
10 KiB
Python
244 lines
10 KiB
Python
"""
|
|
Context Gatherer Unit Tests
|
|
============================
|
|
Phase 5.2.1: 日誌清洗模組測試
|
|
|
|
Gate 2 Checkpoint: 驗證 ERROR Only 過濾邏輯
|
|
- 確保餵給 Ollama 的是純淨的戰訊，不含雜訊
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from src.services.context_gatherer import LogLevelFilter
|
|
|
|
|
|
class TestLogLevelFilter:
|
|
"""LogLevelFilter 單元測試 - ERROR Only 原則驗證"""
|
|
|
|
# =========================================================================
|
|
# 測試案例 1: 禁止的日誌等級 (必須過濾)
|
|
# =========================================================================
|
|
|
|
@pytest.mark.parametrize("line", [
|
|
"[DEBUG] Starting application initialization",
|
|
"[INFO] Server listening on port 8080",
|
|
"[TRACE] Request ID: abc123 processing",
|
|
"[VERBOSE] Memory allocation details",
|
|
"DEBUG: Connection pool initialized",
|
|
"INFO: Health check passed",
|
|
"TRACE: Stack trace dump",
|
|
'level=DEBUG msg="Processing request"',
|
|
'level="INFO" service=api status=healthy',
|
|
'level=info component="scheduler"',
|
|
])
|
|
def test_forbidden_levels_are_filtered(self, line: str):
|
|
"""禁止等級 (DEBUG/INFO/TRACE/VERBOSE) 必須被過濾"""
|
|
assert LogLevelFilter.is_allowed(line) is False, f"Should filter: {line}"
|
|
|
|
# =========================================================================
|
|
# 測試案例 2: 允許的日誌等級 (必須保留)
|
|
# =========================================================================
|
|
|
|
@pytest.mark.parametrize("line", [
|
|
"[ERROR] Database connection failed",
|
|
"[FATAL] Out of memory, shutting down",
|
|
"[CRITICAL] SSL certificate expired",
|
|
"[WARN] High CPU usage detected (95%)",
|
|
"[WARNING] Disk space low on /var/log",
|
|
"ERROR: Unable to connect to Redis",
|
|
"FATAL: Unrecoverable state",
|
|
"CRITICAL: Data corruption detected",
|
|
"WARN: Response time degraded",
|
|
"WARNING: Connection pool exhausted",
|
|
'level=ERROR msg="Request failed"',
|
|
'level="CRITICAL" service=db error="timeout"',
|
|
'level=warning component="cache" status=degraded',
|
|
])
|
|
def test_allowed_levels_are_preserved(self, line: str):
|
|
"""允許等級 (ERROR/FATAL/CRITICAL/WARN/WARNING) 必須保留"""
|
|
assert LogLevelFilter.is_allowed(line) is True, f"Should preserve: {line}"
|
|
|
|
# =========================================================================
|
|
# 測試案例 3: Stacktrace 保留
|
|
# =========================================================================
|
|
|
|
@pytest.mark.parametrize("line", [
|
|
"Traceback (most recent call last):",
|
|
' File "/app/main.py", line 42, in handle_request',
|
|
" at com.example.Service.process(Service.java:123)",
|
|
" at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)",
|
|
"panic: runtime error: index out of range",
|
|
" 0: 0x7fff5fbff8c0 main.main+0x20",
|
|
])
|
|
def test_stacktrace_lines_are_preserved(self, line: str):
|
|
"""Stacktrace 行必須保留 (包括 Python/Java/Go)"""
|
|
assert LogLevelFilter.is_allowed(line) is True, f"Should preserve stacktrace: {line}"
|
|
|
|
# =========================================================================
|
|
# 測試案例 4: K8s 事件格式
|
|
# =========================================================================
|
|
|
|
@pytest.mark.parametrize("line", [
|
|
"Warning BackOff 2m30s kubelet Back-off restarting failed container",
|
|
"Error Failed 5m kubelet Error: ImagePullBackOff",
|
|
])
|
|
def test_k8s_warning_error_events_preserved(self, line: str):
|
|
"""K8s Warning/Error 事件必須保留"""
|
|
assert LogLevelFilter.is_allowed(line) is True, f"Should preserve K8s event: {line}"
|
|
|
|
@pytest.mark.parametrize("line", [
|
|
"Normal Scheduled 10m default-scheduler Successfully assigned",
|
|
"Normal Pulled 8m kubelet Container image pulled",
|
|
])
|
|
def test_k8s_normal_events_filtered(self, line: str):
|
|
"""K8s Normal 事件應該被過濾"""
|
|
assert LogLevelFilter.is_allowed(line) is False, f"Should filter K8s Normal: {line}"
|
|
|
|
# =========================================================================
|
|
# 測試案例 5: 空行與邊界情況
|
|
# =========================================================================
|
|
|
|
@pytest.mark.parametrize("line", [
|
|
"",
|
|
" ",
|
|
"\t\t",
|
|
])
|
|
def test_empty_lines_are_filtered(self, line: str):
|
|
"""空行必須被過濾"""
|
|
assert LogLevelFilter.is_allowed(line) is False
|
|
|
|
# =========================================================================
|
|
# 測試案例 6: 完整日誌過濾 (多行)
|
|
# =========================================================================
|
|
|
|
def test_filter_logs_multiline(self):
|
|
"""測試多行日誌過濾 - ERROR Only 原則"""
|
|
raw_logs = """
|
|
[INFO] Application started successfully
|
|
[DEBUG] Loading configuration from /etc/app/config.yaml
|
|
[INFO] Connected to database
|
|
[ERROR] Failed to connect to Redis: Connection refused
|
|
[INFO] Retrying connection...
|
|
[ERROR] Redis connection failed after 3 retries
|
|
Traceback (most recent call last):
|
|
File "/app/redis_client.py", line 45, in connect
|
|
raise ConnectionError("Unable to connect")
|
|
[DEBUG] Cleanup initiated
|
|
[WARN] Memory usage high: 85%
|
|
[INFO] Health check passed
|
|
[CRITICAL] Service degraded, entering maintenance mode
|
|
""".strip()
|
|
|
|
filtered = LogLevelFilter.filter_logs(raw_logs)
|
|
_lines = [line for line in filtered.split("\n") if line.strip()]
|
|
|
|
# 驗證: 只有 ERROR/WARN/CRITICAL 和 Stacktrace 被保留
|
|
assert "[INFO]" not in filtered, "INFO should be filtered"
|
|
assert "[DEBUG]" not in filtered, "DEBUG should be filtered"
|
|
assert "[ERROR] Failed to connect to Redis" in filtered
|
|
assert "[ERROR] Redis connection failed" in filtered
|
|
assert "Traceback (most recent call last):" in filtered
|
|
assert "[WARN] Memory usage high" in filtered
|
|
assert "[CRITICAL] Service degraded" in filtered
|
|
|
|
# 計算過濾效果
|
|
stats = LogLevelFilter.get_filter_stats(raw_logs, filtered)
|
|
assert stats["filtered_lines"] < stats["original_lines"]
|
|
assert stats["removal_rate_percent"] > 0
|
|
|
|
def test_filter_stats_calculation(self):
|
|
"""測試過濾統計計算"""
|
|
original = "[INFO] line1\n[ERROR] line2\n[DEBUG] line3"
|
|
filtered = "[ERROR] line2"
|
|
|
|
stats = LogLevelFilter.get_filter_stats(original, filtered)
|
|
|
|
assert stats["original_lines"] == 3
|
|
assert stats["filtered_lines"] == 1
|
|
assert stats["removed_lines"] == 2
|
|
assert stats["removal_rate_percent"] == pytest.approx(66.7, rel=0.1)
|
|
|
|
# =========================================================================
|
|
# 測試案例 7: 真實 K8s Pod 日誌模擬
|
|
# =========================================================================
|
|
|
|
def test_real_world_k8s_pod_logs(self):
|
|
"""模擬真實 K8s Pod 日誌 - 驗證雜訊過濾效果"""
|
|
# 模擬 Harbor Core Pod 崩潰日誌
|
|
k8s_logs = """
|
|
2024-03-21T10:15:23.456Z INFO [harbor.core] Starting Harbor Core v2.9.0
|
|
2024-03-21T10:15:24.789Z DEBUG [harbor.core.db] Initializing database connection pool
|
|
2024-03-21T10:15:25.123Z INFO [harbor.core.db] Connected to PostgreSQL
|
|
2024-03-21T10:15:26.456Z DEBUG [harbor.core.cache] Redis client initialized
|
|
2024-03-21T10:15:27.789Z INFO [harbor.core.api] HTTP server listening on :8080
|
|
2024-03-21T10:16:45.123Z ERROR [harbor.core.db] Connection lost to PostgreSQL
|
|
2024-03-21T10:16:45.456Z FATAL [harbor.core] Database connection unrecoverable
|
|
Traceback (most recent call last):
|
|
File "/harbor/core/db.py", line 234, in connect
|
|
raise DatabaseConnectionError("Max retries exceeded")
|
|
2024-03-21T10:16:46.789Z INFO [harbor.core] Graceful shutdown initiated
|
|
2024-03-21T10:16:47.123Z DEBUG [harbor.core] Cleanup completed
|
|
""".strip()
|
|
|
|
filtered = LogLevelFilter.filter_logs(k8s_logs)
|
|
stats = LogLevelFilter.get_filter_stats(k8s_logs, filtered)
|
|
|
|
# 驗證: 只保留 ERROR, FATAL 和 Stacktrace
|
|
assert "ERROR" in filtered
|
|
assert "FATAL" in filtered
|
|
assert "Traceback" in filtered
|
|
assert "INFO" not in filtered.replace("Co", "") # 避免誤判
|
|
assert "DEBUG" not in filtered
|
|
|
|
# 驗證: 過濾率應該很高 (約 60-70%)
|
|
assert stats["removal_rate_percent"] > 50, f"Should filter >50%, got {stats['removal_rate_percent']}%"
|
|
|
|
print("\n📊 K8s Log Filter Stats:")
|
|
print(f" Original: {stats['original_lines']} lines")
|
|
print(f" Filtered: {stats['filtered_lines']} lines")
|
|
print(f" Removed: {stats['removed_lines']} lines ({stats['removal_rate_percent']}%)")
|
|
print(f"\n✅ 純淨戰訊 (ERROR Only):\n{filtered}")
|
|
|
|
|
|
# =============================================================================
|
|
# CLI 測試入口
|
|
# =============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
# 快速驗證測試
|
|
print("=" * 60)
|
|
print("Phase 5.2.1 - Context Gatherer Unit Tests")
|
|
print("Gate 2 Checkpoint: ERROR Only 過濾邏輯驗證")
|
|
print("=" * 60)
|
|
|
|
test = TestLogLevelFilter()
|
|
|
|
# 執行關鍵測試
|
|
print("\n🔍 測試 1: 禁止等級過濾...")
|
|
for line in [
|
|
"[DEBUG] test", "[INFO] test", "[TRACE] test",
|
|
"level=DEBUG msg=test", "INFO: application started",
|
|
]:
|
|
result = LogLevelFilter.is_allowed(line)
|
|
status = "❌ 過濾" if not result else "⚠️ 錯誤保留"
|
|
print(f" {status}: {line[:50]}")
|
|
|
|
print("\n🔍 測試 2: 允許等級保留...")
|
|
for line in [
|
|
"[ERROR] Database connection failed",
|
|
"[FATAL] Out of memory",
|
|
"[CRITICAL] SSL expired",
|
|
"[WARN] High CPU",
|
|
"[WARNING] Disk low",
|
|
]:
|
|
result = LogLevelFilter.is_allowed(line)
|
|
status = "✅ 保留" if result else "⚠️ 錯誤過濾"
|
|
print(f" {status}: {line[:50]}")
|
|
|
|
print("\n🔍 測試 3: 多行日誌過濾效果...")
|
|
test.test_real_world_k8s_pod_logs()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✅ Gate 2 Checkpoint: ERROR Only 過濾邏輯驗證完成")
|
|
print("=" * 60)
|