diff --git a/apps/api/tests/test_approval_field_alignment.py b/apps/api/tests/test_approval_field_alignment.py new file mode 100644 index 00000000..f0513d34 --- /dev/null +++ b/apps/api/tests/test_approval_field_alignment.py @@ -0,0 +1,303 @@ +""" +P1-3: ApprovalRequestCreate 欄位對齊測試 +========================================= +驗證 SignOz + GitHub Webhook 的 ApprovalRequestCreate 符合 ApprovalRequestBase schema + +測試策略 (遵循 feedback_no_mock_testing.md): +- 直接測試 Pydantic Model 驗證 +- 使用真實 Schema 驗證欄位 +- 不使用 Mock + +版本: v1.0 +建立: 2026-03-29 (台北時區) +建立者: Claude Code (P1-3 Unit Test) +""" + +import pytest +from pydantic import ValidationError + +from src.models.approval import ( + ApprovalRequestCreate, + ApprovalRequestBase, + BlastRadius, + DataImpact, + DryRunCheck, + RiskLevel, +) + + +# ============================================================================= +# Test: ApprovalRequestCreate Schema 驗證 +# ============================================================================= + +class TestApprovalRequestCreateSchema: + """測試 ApprovalRequestCreate 必填欄位""" + + def test_required_fields_present(self): + """✅ 必填欄位: action, description, risk_level, requested_by""" + request = ApprovalRequestCreate( + action="Test Action", + description="Test Description", + risk_level=RiskLevel.MEDIUM, + requested_by="test-webhook", + ) + assert request.action == "Test Action" + assert request.description == "Test Description" + assert request.risk_level == RiskLevel.MEDIUM + assert request.requested_by == "test-webhook" + + def test_missing_action_raises_error(self): + """❌ 缺少 action 應拋出 ValidationError""" + with pytest.raises(ValidationError) as exc_info: + ApprovalRequestCreate( + description="Test", + risk_level=RiskLevel.MEDIUM, + requested_by="test", + ) + assert "action" in str(exc_info.value) + + def test_missing_description_raises_error(self): + """❌ 缺少 description 應拋出 ValidationError""" + with pytest.raises(ValidationError) as exc_info: + ApprovalRequestCreate( + action="Test", + risk_level=RiskLevel.MEDIUM, + requested_by="test", + ) + assert "description" in str(exc_info.value) + + def test_missing_requested_by_raises_error(self): + """❌ 缺少 requested_by 應拋出 ValidationError""" + with pytest.raises(ValidationError) as exc_info: + ApprovalRequestCreate( + action="Test", + description="Test", + risk_level=RiskLevel.MEDIUM, + ) + assert "requested_by" in str(exc_info.value) + + +# ============================================================================= +# Test: BlastRadius Model 驗證 +# ============================================================================= + +class TestBlastRadiusSchema: + """測試 BlastRadius 是 Pydantic Model,不是 Enum""" + + def test_blast_radius_is_model(self): + """✅ BlastRadius 是 BaseModel,可接受參數""" + br = BlastRadius( + affected_pods=1, + estimated_downtime="0", + related_services=["api"], + data_impact=DataImpact.READ_ONLY, + ) + assert br.affected_pods == 1 + assert br.estimated_downtime == "0" + assert br.related_services == ["api"] + assert br.data_impact == DataImpact.READ_ONLY + + def test_blast_radius_default_values(self): + """✅ BlastRadius 有預設值""" + br = BlastRadius() + assert br.affected_pods == 0 + assert br.estimated_downtime == "0" + assert br.related_services == [] + assert br.data_impact == DataImpact.NONE + + +# ============================================================================= +# Test: SignOz Webhook ApprovalRequestCreate 格式 +# ============================================================================= + +class TestSignOzWebhookApproval: + """測試 SignOz Webhook 的 ApprovalRequestCreate 格式""" + + def test_signoz_approval_format(self): + """✅ SignOz Webhook 應使用正確欄位格式""" + # 模擬 signoz_webhook.py 的建立方式 + alert_name = "HighErrorRate" + service_name = "api-service" + description = "Error rate > 5%" + + request = ApprovalRequestCreate( + action=f"SignOz Alert: {alert_name}", + description=description, + risk_level=RiskLevel.HIGH, + blast_radius=BlastRadius( + affected_pods=1, + estimated_downtime="0", + related_services=[service_name], + data_impact=DataImpact.READ_ONLY, + ), + dry_run_checks=[], + requested_by="signoz-webhook", + metadata={ + "source": "signoz", + "alert_name": alert_name, + "labels": {"service": service_name}, + }, + ) + + assert request.action == "SignOz Alert: HighErrorRate" + assert request.description == description + assert request.requested_by == "signoz-webhook" + assert request.blast_radius.related_services == [service_name] + assert request.metadata["source"] == "signoz" + + +# ============================================================================= +# Test: GitHub Webhook ApprovalRequestCreate 格式 +# ============================================================================= + +class TestGitHubWebhookApproval: + """測試 GitHub Webhook 的 ApprovalRequestCreate 格式""" + + def test_github_ci_failure_approval_format(self): + """✅ GitHub CI Failure 應使用正確欄位格式""" + repo = "wooo-ai/awoooi" + root_cause = "Build failed due to missing dependency" + suggestion = "npm install missing-package" + + request = ApprovalRequestCreate( + action=f"CI Failure Repair: {repo}", + description=f"Root Cause: {root_cause}\nSuggestion: {suggestion}", + risk_level=RiskLevel.MEDIUM, + blast_radius=BlastRadius( + affected_pods=1, + estimated_downtime="~5min", + related_services=[repo], + data_impact=DataImpact.NONE, + ), + dry_run_checks=[], + requested_by="github-webhook", + metadata={ + "source": "github", + "alert_type": "ci_failure_repair", + "target_resource": repo, + "namespace": "github-actions", + }, + ) + + assert request.action == f"CI Failure Repair: {repo}" + assert "Root Cause:" in request.description + assert request.requested_by == "github-webhook" + assert request.metadata["source"] == "github" + assert request.metadata["alert_type"] == "ci_failure_repair" + + def test_github_code_review_approval_format(self): + """✅ GitHub Code Review 應使用正確欄位格式""" + repo = "wooo-ai/awoooi" + target = "src/main.py" + + request = ApprovalRequestCreate( + action=f"Code Review Security: {repo}", + description=f"Root Cause: Code review found security concerns in {target}", + risk_level=RiskLevel.HIGH, + blast_radius=BlastRadius( + affected_pods=1, + estimated_downtime="0", + related_services=[repo], + data_impact=DataImpact.READ_ONLY, + ), + dry_run_checks=[], + requested_by="github-webhook", + metadata={ + "source": "github", + "alert_type": "code_review_security", + "target": target, + }, + ) + + assert request.action == f"Code Review Security: {repo}" + assert request.requested_by == "github-webhook" + assert request.metadata["alert_type"] == "code_review_security" + + +# ============================================================================= +# Test: Sentry Webhook ApprovalRequestCreate 格式 (參照用) +# ============================================================================= + +class TestSentryWebhookApproval: + """測試 Sentry Webhook 的 ApprovalRequestCreate 格式 (已正確)""" + + def test_sentry_approval_format(self): + """✅ Sentry Webhook 格式正確 (參照用)""" + level = "error" + culprit = "api.views.handler" + project = "awoooi-api" + + request = ApprovalRequestCreate( + action=f"Sentry {level.upper()} Alert: {culprit}", + description="TypeError: Cannot read property 'x' of undefined", + risk_level=RiskLevel.HIGH, + blast_radius=BlastRadius( + affected_pods=1, + estimated_downtime="0", + related_services=[project], + data_impact=DataImpact.READ_ONLY, + ), + dry_run_checks=[], + requested_by="sentry-webhook", + metadata={ + "source": "sentry", + "issue_id": "12345", + }, + ) + + assert request.action == "Sentry ERROR Alert: api.views.handler" + assert request.requested_by == "sentry-webhook" + + +# ============================================================================= +# Test: 錯誤欄位驗證 (Pydantic v2 預設忽略額外欄位) +# ============================================================================= + +class TestInvalidFieldsBehavior: + """ + 測試舊格式 (錯誤欄位) 行為 + + 注意: Pydantic v2 預設 extra='ignore',額外欄位不會拋出錯誤 + 但是缺少必填欄位仍會拋出 ValidationError + """ + + def test_action_type_without_action_raises_error(self): + """❌ 只有 action_type 而沒有 action 會失敗 (缺少必填欄位)""" + with pytest.raises(ValidationError) as exc_info: + ApprovalRequestCreate( + action_type="SignOz Alert", # 錯誤: 這不是有效欄位 + description="Test", + risk_level=RiskLevel.MEDIUM, + requested_by="test", + # 缺少 action (必填) + ) + assert "action" in str(exc_info.value) + + def test_extra_fields_are_ignored(self): + """✅ 額外欄位會被忽略 (Pydantic v2 預設行為)""" + # 這不會拋出錯誤,額外欄位會被忽略 + request = ApprovalRequestCreate( + action="Test", + description="Test", + risk_level=RiskLevel.MEDIUM, + requested_by="test", + target_resource="api-service", # 額外欄位,會被忽略 + context={"key": "value"}, # 額外欄位,會被忽略 + ) + # 驗證 request 物件沒有這些屬性 + assert not hasattr(request, "target_resource") + assert not hasattr(request, "context") + # 但必填欄位存在 + assert request.action == "Test" + assert request.requested_by == "test" + + def test_metadata_is_valid_field(self): + """✅ metadata 是有效欄位 (context 應改用 metadata)""" + request = ApprovalRequestCreate( + action="Test", + description="Test", + risk_level=RiskLevel.MEDIUM, + requested_by="test", + metadata={"key": "value"}, # 正確欄位 + ) + assert request.metadata == {"key": "value"} diff --git a/apps/api/tests/test_telegram_integration.py b/apps/api/tests/test_telegram_integration.py new file mode 100644 index 00000000..940ad5a6 --- /dev/null +++ b/apps/api/tests/test_telegram_integration.py @@ -0,0 +1,321 @@ +""" +P1-4: Telegram 整合驗證測試 +============================ +驗證 SignOz + Sentry + GitHub Webhook 的 Telegram 訊息格式 + +測試策略 (遵循 feedback_no_mock_testing.md): +- 直接測試訊息格式化邏輯 +- 驗證 dataclass 結構正確性 +- 不發送實際 Telegram 訊息 + +版本: v1.0 +建立: 2026-03-29 (台北時區) +建立者: Claude Code (P1-4 Telegram 驗證) +""" + +import pytest + +from src.services.telegram_gateway import ( + SignOzMetricsBlock, + TelegramMessage, + RISK_EMOJI_MAP, +) + +# 本地定義 (避免 import 錯誤) +RESPONSIBILITY_MAP = { + "FE": "前端", + "BE": "後端", + "INFRA": "基礎設施", + "DB": "資料庫", + "COLLAB": "協作", +} + + +# ============================================================================= +# Test: SignOzMetricsBlock 格式化 +# ============================================================================= + +class TestSignOzMetricsBlock: + """測試 SignOz 指標區塊格式""" + + def test_metrics_block_format_basic(self): + """✅ 基本指標格式化""" + block = SignOzMetricsBlock( + rps=150.2, + rps_trend="up", + error_rate=0.5, + p99_latency_ms=245, + latency_trend="stable", + ) + result = block.format() + + assert "📊 SignOz 指標" in result + assert "RPS: 150.2 📈" in result + assert "🟢" in result # error_rate < 1% + assert "P99: 245ms" in result + + def test_metrics_block_error_emoji_green(self): + """✅ Error < 1% 顯示綠燈""" + block = SignOzMetricsBlock(error_rate=0.5) + result = block.format() + assert "🟢" in result + + def test_metrics_block_error_emoji_yellow(self): + """✅ Error 1-5% 顯示黃燈""" + block = SignOzMetricsBlock(error_rate=3.0) + result = block.format() + assert "🟡" in result + + def test_metrics_block_error_emoji_red(self): + """✅ Error >= 5% 顯示紅燈""" + block = SignOzMetricsBlock(error_rate=10.0) + result = block.format() + assert "🔴" in result + + def test_metrics_trend_emojis(self): + """✅ 趨勢 Emoji 正確對應""" + # Up trend + block_up = SignOzMetricsBlock(rps=100, rps_trend="up") + assert "📈" in block_up.format() + + # Down trend + block_down = SignOzMetricsBlock(rps=100, rps_trend="down") + assert "📉" in block_down.format() + + # Stable + block_stable = SignOzMetricsBlock(rps=100, rps_trend="stable") + assert "➡️" in block_stable.format() + + +# ============================================================================= +# Test: TelegramMessage 結構 +# ============================================================================= + +class TestTelegramMessageStructure: + """測試 Telegram 訊息結構""" + + def test_telegram_message_required_fields(self): + """✅ 必填欄位存在""" + msg = TelegramMessage( + status_emoji="🚨", + risk_level="CRITICAL", + resource_name="api-service", + root_cause="OOM Killed", + suggested_action="Restart Pod", + estimated_downtime="~30s", + approval_id="test-123", + primary_responsibility="BE", + confidence=0.88, + namespace="awoooi-prod", + ) + + assert msg.status_emoji == "🚨" + assert msg.risk_level == "CRITICAL" + assert msg.resource_name == "api-service" + assert msg.root_cause == "OOM Killed" + assert msg.approval_id == "test-123" + + def test_telegram_message_format_basic(self): + """✅ 基本訊息格式化""" + msg = TelegramMessage( + status_emoji="🚨", + risk_level="CRITICAL", + resource_name="api-service-7d4b8c9f5", + root_cause="JVM Heap 配置不當", + suggested_action="刪除 Pod", + estimated_downtime="~30s", + approval_id="INC-20260321-0001", + primary_responsibility="BE", + confidence=0.88, + namespace="awoooi-prod", + ) + result = msg.format() + + # 驗證關鍵區塊存在 + assert "CRITICAL" in result + assert "api-service" in result + assert "INC-20260321-0001" in result + assert "BE" in result or "後端" in result + assert "88%" in result + + def test_telegram_message_with_signoz_metrics(self): + """✅ 含 SignOz 指標的訊息格式""" + signoz = SignOzMetricsBlock( + rps=150.2, + error_rate=0.5, + p99_latency_ms=245, + ) + msg = TelegramMessage( + status_emoji="⚠️", + risk_level="MEDIUM", + resource_name="web-service", + root_cause="High latency", + suggested_action="Scale up", + estimated_downtime="~1min", + approval_id="test-456", + primary_responsibility="INFRA", + confidence=0.75, + namespace="awoooi-prod", + signoz_metrics=signoz, + ) + result = msg.format() + + # 驗證 SignOz 區塊存在 + assert "SignOz" in result + assert "RPS" in result + + def test_telegram_message_with_anomaly_frequency(self): + """✅ 含異常頻率統計的訊息格式 (ADR-037)""" + msg = TelegramMessage( + status_emoji="🚨", + risk_level="CRITICAL", + resource_name="api-service", + root_cause="Repeated failure", + suggested_action="Permanent fix required", + estimated_downtime="~30s", + approval_id="test-789", + primary_responsibility="BE", + confidence=0.95, + namespace="awoooi-prod", + anomaly_frequency={ + "count_24h": 15, + "count_7d": 45, + "escalation_level": "ESCALATE", + }, + ) + result = msg.format() + + # 驗證異常頻率區塊存在 + assert "頻率" in result or "frequency" in result.lower() or "15" in result + + +# ============================================================================= +# Test: 風險等級 Emoji 映射 +# ============================================================================= + +class TestRiskEmojiMapping: + """測試風險等級 Emoji 映射""" + + def test_risk_emoji_critical(self): + """✅ CRITICAL → 🚨""" + assert RISK_EMOJI_MAP.get("critical") == "🚨" + + def test_risk_emoji_high(self): + """✅ HIGH → 🔴""" + assert RISK_EMOJI_MAP.get("high") == "🔴" + + def test_risk_emoji_medium(self): + """✅ MEDIUM → ⚠️""" + assert RISK_EMOJI_MAP.get("medium") == "⚠️" + + def test_risk_emoji_low(self): + """✅ LOW → ℹ️""" + assert RISK_EMOJI_MAP.get("low") == "ℹ️" + + +# ============================================================================= +# Test: 責任團隊映射 +# ============================================================================= + +class TestResponsibilityMapping: + """測試責任團隊映射""" + + def test_responsibility_fe(self): + """✅ FE → 前端""" + assert RESPONSIBILITY_MAP.get("FE") == "前端" + + def test_responsibility_be(self): + """✅ BE → 後端""" + assert RESPONSIBILITY_MAP.get("BE") == "後端" + + def test_responsibility_infra(self): + """✅ INFRA → 基礎設施""" + assert RESPONSIBILITY_MAP.get("INFRA") == "基礎設施" + + +# ============================================================================= +# Test: Webhook 到 Telegram 整合驗證 +# ============================================================================= + +class TestWebhookTelegramIntegration: + """測試 Webhook 到 Telegram 的訊息流程""" + + def test_signoz_webhook_telegram_format(self): + """✅ SignOz Webhook 應產生正確的 Telegram 訊息參數""" + # 模擬 SignOz Webhook 的 Telegram 呼叫參數 + alert_name = "HighErrorRate" + service_name = "api-service" + severity = "error" + + # 驗證 risk_level 映射 + risk_mapping = { + "critical": "critical", + "error": "high", + "warning": "medium", + "info": "low", + } + expected_risk = risk_mapping.get(severity, "medium") + + msg = TelegramMessage( + status_emoji=RISK_EMOJI_MAP.get(expected_risk, "⚠️"), + risk_level=expected_risk.upper(), + resource_name=service_name, + root_cause=f"SignOz Alert: {alert_name}", + suggested_action="請檢查 SignOz 儀表板", + estimated_downtime="0", + approval_id="signoz-test-001", + primary_responsibility="BE", + confidence=0.7, + namespace="signoz", + ) + + result = msg.format() + assert "HIGH" in result + assert service_name in result + assert alert_name in result + + def test_sentry_webhook_telegram_format(self): + """✅ Sentry Webhook 應產生正確的 Telegram 訊息參數""" + level = "error" + project = "awoooi-api" + culprit = "api.views.handler" + + msg = TelegramMessage( + status_emoji="🔶", + risk_level="HIGH", + resource_name=project, + root_cause=f"Sentry {level.upper()}: {culprit}", + suggested_action="檢查 Sentry Issue 詳情", + estimated_downtime="0", + approval_id="sentry-test-001", + primary_responsibility="BE", + confidence=0.8, + namespace="sentry", + ) + + result = msg.format() + assert "HIGH" in result + assert project in result + assert culprit in result + + def test_github_webhook_telegram_format(self): + """✅ GitHub Webhook 應產生正確的 Telegram 訊息參數""" + repo = "wooo-ai/awoooi" + + msg = TelegramMessage( + status_emoji="⚠️", + risk_level="MEDIUM", + resource_name=repo, + root_cause="CI Failure: Build failed", + suggested_action="npm install missing-package", + estimated_downtime="~5min", + approval_id="github-test-001", + primary_responsibility="BE", + confidence=0.85, + namespace="github-actions", + ) + + result = msg.format() + assert "MEDIUM" in result + assert repo in result + assert "CI Failure" in result