d89f0520f9
- 自動修復 import 排序、unused imports - 手動修復 raise from、isinstance union、unused variable - scripts/ 暫時保留 (非 CI 阻擋) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
391 lines
12 KiB
Python
Executable File
391 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
AWOOOI Alert Chain Smoke Test (Wave A.6)
|
|
=========================================
|
|
E2E 驗證告警鏈路完整性
|
|
|
|
功能:
|
|
1. 發送測試告警到 Sentry/SignOz Webhook
|
|
2. 驗證 Telegram 通知發送
|
|
3. 檢查 Prometheus Metrics
|
|
4. 驗證 Incident 建立
|
|
|
|
用法:
|
|
cd apps/api
|
|
python -m scripts.alert_chain_smoke_test
|
|
|
|
# 指定來源
|
|
python -m scripts.alert_chain_smoke_test --source sentry
|
|
python -m scripts.alert_chain_smoke_test --source signoz
|
|
python -m scripts.alert_chain_smoke_test --source all
|
|
|
|
環境變數:
|
|
AWOOOI_API_URL: API 端點 (預設 http://localhost:8000)
|
|
AWOOOI_METRICS_URL: Metrics 端點 (預設 http://localhost:8000/metrics)
|
|
|
|
Author: Claude Code
|
|
Date: 2026-03-29
|
|
ADR: ADR-037 (監控增強架構)
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
# Add src to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import httpx
|
|
|
|
# =============================================================================
|
|
# Config (從環境變數讀取，符合首席架構師審查 P0 要求)
|
|
# =============================================================================
|
|
|
|
API_URL = os.environ.get("AWOOOI_API_URL", "http://localhost:8000")
|
|
METRICS_URL = os.environ.get("AWOOOI_METRICS_URL", f"{API_URL}/metrics")
|
|
|
|
# Webhook 端點
|
|
SENTRY_WEBHOOK = f"{API_URL}/api/v1/webhooks/sentry/issue"
|
|
SIGNOZ_WEBHOOK = f"{API_URL}/api/v1/webhooks/signoz/alert"
|
|
|
|
# =============================================================================
|
|
# Test Payloads
|
|
# =============================================================================
|
|
|
|
SENTRY_TEST_PAYLOAD = {
|
|
"action": "triggered",
|
|
"data": {
|
|
"issue": {
|
|
"id": f"smoke-test-{int(time.time())}",
|
|
"title": "[Smoke Test] Alert Chain Verification",
|
|
"culprit": "alert_chain_smoke_test.py",
|
|
"level": "error",
|
|
"firstSeen": datetime.now().isoformat(),
|
|
"count": 1,
|
|
"project": {"slug": "awoooi-api"},
|
|
},
|
|
"event": {
|
|
"message": "Smoke test alert - verifying alert chain E2E",
|
|
"platform": "python",
|
|
"tags": [{"key": "smoke_test", "value": "true"}],
|
|
},
|
|
},
|
|
}
|
|
|
|
SIGNOZ_TEST_PAYLOAD = {
|
|
"alertname": "SmokeTestAlert",
|
|
"status": "firing",
|
|
"labels": {
|
|
"severity": "warning",
|
|
"service_name": "smoke-test",
|
|
"namespace": "awoooi-test",
|
|
"source": "signoz",
|
|
},
|
|
"annotations": {
|
|
"summary": "[Smoke Test] SignOz Alert Chain Verification",
|
|
"description": "This is a smoke test alert to verify the alert chain E2E",
|
|
},
|
|
"startsAt": datetime.now().isoformat(),
|
|
}
|
|
|
|
|
|
# =============================================================================
|
|
# Terminal Output Helpers
|
|
# =============================================================================
|
|
|
|
class Colors:
|
|
"""ANSI Color Codes"""
|
|
HEADER = "\033[95m"
|
|
BLUE = "\033[94m"
|
|
CYAN = "\033[96m"
|
|
GREEN = "\033[92m"
|
|
YELLOW = "\033[93m"
|
|
RED = "\033[91m"
|
|
ENDC = "\033[0m"
|
|
BOLD = "\033[1m"
|
|
DIM = "\033[2m"
|
|
|
|
|
|
def print_banner():
|
|
"""Print banner"""
|
|
print(f"""
|
|
{Colors.CYAN}{Colors.BOLD}
|
|
╔═══════════════════════════════════════════════════════╗
|
|
║ AWOOOI Alert Chain Smoke Test (ADR-037 Wave A.6) ║
|
|
╚═══════════════════════════════════════════════════════╝
|
|
{Colors.ENDC}
|
|
{Colors.DIM} API URL: {API_URL}
|
|
Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (Taipei){Colors.ENDC}
|
|
""")
|
|
|
|
|
|
def print_section(title: str):
|
|
"""Print section header"""
|
|
print(f"\n{Colors.BLUE}{Colors.BOLD}▶ {title}{Colors.ENDC}")
|
|
print(f"{Colors.DIM}{'─' * 55}{Colors.ENDC}")
|
|
|
|
|
|
def print_result(success: bool, message: str, details: str = ""):
|
|
"""Print test result"""
|
|
if success:
|
|
print(f" {Colors.GREEN}✓ {message}{Colors.ENDC}")
|
|
else:
|
|
print(f" {Colors.RED}✗ {message}{Colors.ENDC}")
|
|
if details:
|
|
print(f" {Colors.DIM}{details}{Colors.ENDC}")
|
|
|
|
|
|
# =============================================================================
|
|
# Test Functions
|
|
# =============================================================================
|
|
|
|
async def test_sentry_webhook() -> tuple[bool, dict]:
|
|
"""
|
|
測試 Sentry Webhook
|
|
|
|
Returns:
|
|
(success, response_data)
|
|
"""
|
|
print_section("1. Sentry Webhook Test")
|
|
print(f" {Colors.CYAN}Endpoint:{Colors.ENDC} {SENTRY_WEBHOOK}")
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(
|
|
SENTRY_WEBHOOK,
|
|
json=SENTRY_TEST_PAYLOAD,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
result = response.json()
|
|
success = response.status_code in (200, 202) and result.get("status") in ("accepted", "ok")
|
|
|
|
print_result(
|
|
success,
|
|
f"HTTP {response.status_code}",
|
|
f"Response: {result}",
|
|
)
|
|
|
|
return success, result
|
|
|
|
except httpx.ConnectError:
|
|
print_result(False, "Connection failed", f"Cannot reach {SENTRY_WEBHOOK}")
|
|
return False, {"error": "connection_failed"}
|
|
except Exception as e:
|
|
print_result(False, "Request failed", str(e))
|
|
return False, {"error": str(e)}
|
|
|
|
|
|
async def test_signoz_webhook() -> tuple[bool, dict]:
|
|
"""
|
|
測試 SignOz Webhook
|
|
|
|
Returns:
|
|
(success, response_data)
|
|
"""
|
|
print_section("2. SignOz Webhook Test")
|
|
print(f" {Colors.CYAN}Endpoint:{Colors.ENDC} {SIGNOZ_WEBHOOK}")
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(
|
|
SIGNOZ_WEBHOOK,
|
|
json=SIGNOZ_TEST_PAYLOAD,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
result = response.json()
|
|
success = response.status_code in (200, 202) and result.get("status") in ("accepted", "ok")
|
|
|
|
print_result(
|
|
success,
|
|
f"HTTP {response.status_code}",
|
|
f"Response: {result}",
|
|
)
|
|
|
|
return success, result
|
|
|
|
except httpx.ConnectError:
|
|
print_result(False, "Connection failed", f"Cannot reach {SIGNOZ_WEBHOOK}")
|
|
return False, {"error": "connection_failed"}
|
|
except Exception as e:
|
|
print_result(False, "Request failed", str(e))
|
|
return False, {"error": str(e)}
|
|
|
|
|
|
async def test_metrics() -> tuple[bool, dict]:
|
|
"""
|
|
檢查 Prometheus Metrics
|
|
|
|
驗證 Alert Chain 指標存在:
|
|
- awoooi_alert_chain_healthy
|
|
- awoooi_alert_chain_last_success_timestamp
|
|
"""
|
|
print_section("3. Prometheus Metrics Check")
|
|
print(f" {Colors.CYAN}Endpoint:{Colors.ENDC} {METRICS_URL}")
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(METRICS_URL)
|
|
|
|
if response.status_code != 200:
|
|
print_result(False, f"HTTP {response.status_code}")
|
|
return False, {"error": f"status_code: {response.status_code}"}
|
|
|
|
metrics_text = response.text
|
|
|
|
# 檢查必要指標
|
|
required_metrics = [
|
|
"awoooi_alert_chain_healthy",
|
|
"awoooi_alert_chain_last_success_timestamp",
|
|
"awoooi_webhook_requests_total",
|
|
"awoooi_alert_processed_total",
|
|
]
|
|
|
|
found_metrics = {}
|
|
for metric in required_metrics:
|
|
found = metric in metrics_text
|
|
found_metrics[metric] = found
|
|
print_result(found, f"Metric: {metric}")
|
|
|
|
all_found = all(found_metrics.values())
|
|
return all_found, found_metrics
|
|
|
|
except httpx.ConnectError:
|
|
print_result(False, "Connection failed", f"Cannot reach {METRICS_URL}")
|
|
return False, {"error": "connection_failed"}
|
|
except Exception as e:
|
|
print_result(False, "Request failed", str(e))
|
|
return False, {"error": str(e)}
|
|
|
|
|
|
async def test_health() -> tuple[bool, dict]:
|
|
"""
|
|
檢查 API Health
|
|
"""
|
|
print_section("4. API Health Check")
|
|
health_url = f"{API_URL}/api/v1/health"
|
|
print(f" {Colors.CYAN}Endpoint:{Colors.ENDC} {health_url}")
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(health_url)
|
|
|
|
success = response.status_code == 200
|
|
result = response.json() if success else {}
|
|
|
|
print_result(
|
|
success,
|
|
f"HTTP {response.status_code}",
|
|
f"Status: {result.get('status', 'unknown')}",
|
|
)
|
|
|
|
return success, result
|
|
|
|
except httpx.ConnectError:
|
|
print_result(False, "Connection failed", f"Cannot reach {health_url}")
|
|
return False, {"error": "connection_failed"}
|
|
except Exception as e:
|
|
print_result(False, "Request failed", str(e))
|
|
return False, {"error": str(e)}
|
|
|
|
|
|
# =============================================================================
|
|
# Main
|
|
# =============================================================================
|
|
|
|
async def run_smoke_tests(source: str = "all") -> bool:
|
|
"""
|
|
執行 Smoke Tests
|
|
|
|
Args:
|
|
source: 測試來源 (sentry, signoz, all)
|
|
|
|
Returns:
|
|
bool: 全部通過
|
|
"""
|
|
print_banner()
|
|
|
|
results = []
|
|
|
|
# Health Check
|
|
health_ok, _ = await test_health()
|
|
results.append(("Health", health_ok))
|
|
|
|
if not health_ok:
|
|
print(f"\n{Colors.RED}{Colors.BOLD}✗ API 健康檢查失敗，終止測試{Colors.ENDC}")
|
|
return False
|
|
|
|
# Webhook Tests
|
|
if source in ("sentry", "all"):
|
|
sentry_ok, _ = await test_sentry_webhook()
|
|
results.append(("Sentry Webhook", sentry_ok))
|
|
|
|
if source in ("signoz", "all"):
|
|
signoz_ok, _ = await test_signoz_webhook()
|
|
results.append(("SignOz Webhook", signoz_ok))
|
|
|
|
# 等待背景任務處理
|
|
if source != "none":
|
|
print(f"\n{Colors.DIM} 等待 2 秒讓背景任務處理...{Colors.ENDC}")
|
|
await asyncio.sleep(2)
|
|
|
|
# Metrics Check
|
|
metrics_ok, _ = await test_metrics()
|
|
results.append(("Metrics", metrics_ok))
|
|
|
|
# Summary
|
|
print_section("Summary")
|
|
all_passed = True
|
|
for name, passed in results:
|
|
print_result(passed, name)
|
|
if not passed:
|
|
all_passed = False
|
|
|
|
print(f"\n{'─' * 55}")
|
|
if all_passed:
|
|
print(f"{Colors.GREEN}{Colors.BOLD}✓ All smoke tests passed!{Colors.ENDC}")
|
|
else:
|
|
print(f"{Colors.RED}{Colors.BOLD}✗ Some smoke tests failed!{Colors.ENDC}")
|
|
|
|
return all_passed
|
|
|
|
|
|
def main():
|
|
"""CLI Entry Point"""
|
|
parser = argparse.ArgumentParser(
|
|
description="AWOOOI Alert Chain Smoke Test (ADR-037 Wave A.6)",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
環境變數:
|
|
AWOOOI_API_URL API 端點 (預設 http://localhost:8000)
|
|
AWOOOI_METRICS_URL Metrics 端點 (預設 $AWOOOI_API_URL/metrics)
|
|
|
|
範例:
|
|
python -m scripts.alert_chain_smoke_test
|
|
python -m scripts.alert_chain_smoke_test --source sentry
|
|
AWOOOI_API_URL=http://awoooi-api:8000 python -m scripts.alert_chain_smoke_test
|
|
""",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--source", "-s",
|
|
type=str,
|
|
default="all",
|
|
choices=["sentry", "signoz", "all", "none"],
|
|
help="測試來源 (預設: all, none = 只檢查 health/metrics)",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
success = asyncio.run(run_smoke_tests(args.source))
|
|
sys.exit(0 if success else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|