6f049877fc
- Python: ruff --fix 修復 280 個 lint 錯誤 - lewooogo-core: src/ 目錄未追蹤，導致 CI eslint 失敗 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
244 lines
8.6 KiB
Python
244 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Phase 5 E2E 點火測試 - OpenClaw 全鏈路驗證
|
||
==========================================
|
||
|
||
測試流程:
|
||
1. 發射模擬 K8s 告警到 Webhook
|
||
2. 驗證告警被正確處理
|
||
3. 驗證 ApprovalRecord 被建立
|
||
4. 模擬 Telegram 簽核回調
|
||
5. 驗證執行觸發
|
||
|
||
使用方式:
|
||
python scripts/e2e_openclaw_test.py
|
||
"""
|
||
|
||
import asyncio
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
|
||
def print_header(title: str) -> None:
|
||
"""列印測試標題"""
|
||
print("\n" + "=" * 60)
|
||
print(f" {title}")
|
||
print("=" * 60)
|
||
|
||
|
||
def print_step(step: int, description: str) -> None:
|
||
"""列印測試步驟"""
|
||
print(f"\n🔹 Step {step}: {description}")
|
||
|
||
|
||
def print_success(message: str) -> None:
|
||
"""列印成功訊息"""
|
||
print(f" ✅ {message}")
|
||
|
||
|
||
def print_error(message: str) -> None:
|
||
"""列印錯誤訊息"""
|
||
print(f" ❌ {message}")
|
||
|
||
|
||
def print_info(message: str) -> None:
|
||
"""列印資訊訊息"""
|
||
print(f" ℹ️ {message}")
|
||
|
||
|
||
async def test_phase5_e2e():
|
||
"""Phase 5 E2E 測試"""
|
||
print_header("Phase 5 E2E 點火測試 - OpenClaw 全鏈路驗證")
|
||
print(f"執行時間: {datetime.now().isoformat()}")
|
||
|
||
# =========================================================================
|
||
# Step 1: 測試 LogLevelFilter (日誌清洗)
|
||
# =========================================================================
|
||
print_step(1, "日誌清洗模組 (LogLevelFilter)")
|
||
|
||
try:
|
||
from src.services.context_gatherer import LogLevelFilter
|
||
|
||
# 模擬 K8s 日誌
|
||
raw_logs = """
|
||
2024-03-21T10:15:23.456Z INFO [harbor.core] Starting Harbor Core
|
||
2024-03-21T10:15:24.789Z DEBUG [harbor.core.db] Initializing connection pool
|
||
2024-03-21T10:16:45.123Z ERROR [harbor.core.db] Connection lost to PostgreSQL
|
||
2024-03-21T10:16:45.456Z FATAL [harbor.core] Unrecoverable error
|
||
Traceback (most recent call last):
|
||
File "/harbor/core/db.py", line 234, in connect
|
||
raise DatabaseConnectionError("Max retries exceeded")
|
||
""".strip()
|
||
|
||
filtered = LogLevelFilter.filter_logs(raw_logs)
|
||
stats = LogLevelFilter.get_filter_stats(raw_logs, filtered)
|
||
|
||
# 驗證 DEBUG/INFO 被過濾
|
||
assert "DEBUG" not in filtered, "DEBUG should be filtered"
|
||
assert "INFO" not in filtered.replace("DatabaseConnectionError", ""), "INFO should be filtered"
|
||
assert "ERROR" in filtered, "ERROR should be preserved"
|
||
assert "FATAL" in filtered, "FATAL should be preserved"
|
||
assert "Traceback" in filtered, "Stacktrace should be preserved"
|
||
|
||
print_success(f"日誌清洗成功: {stats['original_lines']} → {stats['filtered_lines']} 行")
|
||
print_success(f"雜訊移除率: {stats['removal_rate_percent']}%")
|
||
|
||
except Exception as e:
|
||
print_error(f"日誌清洗測試失敗: {e}")
|
||
return False
|
||
|
||
# =========================================================================
|
||
# Step 2: 測試 Security Interceptor (白名單 + Nonce)
|
||
# =========================================================================
|
||
print_step(2, "安全攔截器 (Security Interceptor)")
|
||
|
||
try:
|
||
from src.core.config import settings
|
||
from src.services.security_interceptor import (
|
||
TelegramSecurityInterceptor,
|
||
)
|
||
|
||
interceptor = TelegramSecurityInterceptor()
|
||
|
||
# 測試白名單 (假設統帥 ID: 5619078117)
|
||
test_user_id = 5619078117
|
||
|
||
# 檢查白名單配置
|
||
whitelist = settings.OPENCLAW_TG_USER_WHITELIST
|
||
print_info(f"白名單配置: {whitelist}")
|
||
|
||
if whitelist:
|
||
is_whitelisted = interceptor.is_whitelisted(test_user_id)
|
||
if is_whitelisted:
|
||
print_success(f"統帥 ID {test_user_id} 在白名單內")
|
||
else:
|
||
print_info(f"統帥 ID {test_user_id} 不在白名單 (需配置)")
|
||
else:
|
||
print_info("白名單為空 (需在環境變數中配置 OPENCLAW_TG_USER_WHITELIST)")
|
||
|
||
# 測試 Nonce 產生
|
||
nonce = interceptor.generate_callback_nonce("test-approval-123", "approve")
|
||
print_success(f"Nonce 產生成功: {nonce[:30]}...")
|
||
|
||
# 解析 Nonce
|
||
parsed = interceptor.parse_callback_data(nonce)
|
||
assert parsed["action"] == "approve"
|
||
assert parsed["approval_id"] == "test-approval-123"
|
||
print_success("Nonce 解析成功")
|
||
|
||
except Exception as e:
|
||
print_error(f"安全攔截器測試失敗: {e}")
|
||
return False
|
||
|
||
# =========================================================================
|
||
# Step 3: 測試 Telegram Gateway (訊息格式)
|
||
# =========================================================================
|
||
print_step(3, "Telegram Gateway (SOUL.md 訊息格式)")
|
||
|
||
try:
|
||
from src.services.telegram_gateway import RISK_EMOJI_MAP, TelegramMessage
|
||
|
||
# 建立測試訊息
|
||
message = TelegramMessage(
|
||
status_emoji=RISK_EMOJI_MAP["critical"],
|
||
risk_level="CRITICAL",
|
||
resource_name="harbor-core-7d4b8c9f5-xk2m3",
|
||
root_cause="OOMKilled",
|
||
suggested_action="DELETE_POD (重啟 Pod)",
|
||
estimated_downtime="~30s",
|
||
approval_id="test-approval-123",
|
||
)
|
||
|
||
formatted = message.format()
|
||
|
||
# 驗證 SOUL.md 格式
|
||
assert "🚨" in formatted, "Should have critical emoji"
|
||
assert "CRITICAL" in formatted, "Should have risk level"
|
||
assert "harbor-core" in formatted, "Should have resource name"
|
||
assert "OOMKilled" in formatted, "Should have root cause"
|
||
assert "建議" in formatted, "Should have suggestion"
|
||
assert "停機" in formatted, "Should have downtime"
|
||
assert len(formatted) <= 500, f"Should be <= 500 chars, got {len(formatted)}"
|
||
|
||
print_success("SOUL.md 訊息格式驗證通過")
|
||
print_info(f"訊息長度: {len(formatted)} / 500 字元")
|
||
print()
|
||
print(" 📱 訊息預覽:")
|
||
for line in formatted.split("\n"):
|
||
print(f" {line}")
|
||
|
||
except Exception as e:
|
||
print_error(f"Telegram Gateway 測試失敗: {e}")
|
||
return False
|
||
|
||
# =========================================================================
|
||
# Step 4: 測試 OpenClaw 模組載入
|
||
# =========================================================================
|
||
print_step(4, "OpenClaw AI 模組載入")
|
||
|
||
try:
|
||
from src.services.openclaw import OpenClawService, get_openclaw
|
||
|
||
openclaw = get_openclaw()
|
||
assert isinstance(openclaw, OpenClawService)
|
||
print_success("OpenClaw 服務載入成功")
|
||
|
||
# 檢查 AI Fallback 順序
|
||
from src.core.config import settings
|
||
print_info(f"AI Fallback 順序: {settings.AI_FALLBACK_ORDER}")
|
||
print_info(f"預設模型: {settings.OPENCLAW_DEFAULT_MODEL}")
|
||
|
||
except Exception as e:
|
||
print_error(f"OpenClaw 模組載入失敗: {e}")
|
||
return False
|
||
|
||
# =========================================================================
|
||
# Step 5: 測試 Signature 審計欄位
|
||
# =========================================================================
|
||
print_step(5, "Signature 審計欄位 (Telegram 擴充)")
|
||
|
||
try:
|
||
from src.models.approval import Signature, SignatureSource
|
||
|
||
# 建立 Telegram 簽核記錄
|
||
sig = Signature(
|
||
signer_id="tg_5619078117",
|
||
signer_name="統帥",
|
||
comment="Telegram 簽核測試",
|
||
source=SignatureSource.TELEGRAM,
|
||
telegram_user_id=5619078117,
|
||
telegram_message_id=12345,
|
||
)
|
||
|
||
assert sig.source == SignatureSource.TELEGRAM
|
||
assert sig.telegram_user_id == 5619078117
|
||
print_success("Telegram 審計欄位驗證通過")
|
||
print_info(f"簽核來源: {sig.source.value}")
|
||
print_info(f"Telegram User ID: {sig.telegram_user_id}")
|
||
|
||
except Exception as e:
|
||
print_error(f"Signature 審計欄位測試失敗: {e}")
|
||
return False
|
||
|
||
# =========================================================================
|
||
# 測試完成
|
||
# =========================================================================
|
||
print_header("E2E 測試結果")
|
||
print()
|
||
print(" ✅ Step 1: 日誌清洗 (LogLevelFilter) - PASSED")
|
||
print(" ✅ Step 2: 安全攔截器 (Security Interceptor) - PASSED")
|
||
print(" ✅ Step 3: Telegram Gateway (SOUL.md 格式) - PASSED")
|
||
print(" ✅ Step 4: OpenClaw AI 模組載入 - PASSED")
|
||
print(" ✅ Step 5: Signature 審計欄位 - PASSED")
|
||
print()
|
||
print("=" * 60)
|
||
print(" 🎉 Phase 5 E2E 點火測試 - 全數通過！")
|
||
print("=" * 60)
|
||
|
||
return True
|
||
|
||
|
||
if __name__ == "__main__":
|
||
success = asyncio.run(test_phase5_e2e())
|
||
sys.exit(0 if success else 1)
|