wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

fix(api): 修復 34 個 Ruff lint 錯誤

Browse Source 
- 自動修復 import 排序、unused imports
- 手動修復 raise from、isinstance union、unused variable
- scripts/ 暫時保留 (非 CI 阻擋)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-29 15:27:49 +08:00
parent 5f9a6a7e55
commit d89f0520f9
27 changed files with 2538 additions and 1132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

362
apps/api/tests/test_adr030_auto_approve.py
View File

@@ -1,362 +0,0 @@
"""
ADR-030 Auto Approve Policy Tests
=================================
測試自動執行策略服務
版本: v1.0
建立: 2026-03-26 (台北時區)
建立者: Claude Code (ADR-030 Phase 4)
"""
from unittest.mock import MagicMock
import pytest
from src.models.playbook import (
ActionType,
Playbook,
PlaybookStatus,
RepairStep,
RiskLevel,
SymptomPattern,
)
from src.services.auto_approve import (
AutoApproveConfig,
AutoApprovePolicy,
AutoApproveReason,
)
from src.services.playbook_rag import PlaybookMatch
class TestAutoApprovePolicy:
"""AutoApprovePolicy 單元測試"""
@pytest.fixture
def mock_trust_manager(self):
"""建立 mock TrustScoreManager"""
manager = MagicMock()
# 預設信任分數為 5
manager.get_trust_record.return_value = MagicMock(score=5)
return manager
@pytest.fixture
def policy(self, mock_trust_manager):
return AutoApprovePolicy(trust_manager=mock_trust_manager)
def create_proposal_data(
self,
risk_level: str = "low",
confidence: float = 0.95,
action: str = "kubectl rollout restart deployment/test-app -n prod",
) -> dict:
"""建立測試用 proposal_data"""
return {
"risk_level": risk_level,
"confidence": confidence,
"action": action,
}
def create_playbook(
self,
success_count: int = 10,
failure_count: int = 0,
risk_level: RiskLevel = RiskLevel.LOW,
) -> Playbook:
"""建立測試用 Playbook"""
return Playbook(
playbook_id="PB-TEST-001",
name="Test Playbook",
description="For testing",
status=PlaybookStatus.APPROVED,
symptom_pattern=SymptomPattern(
alert_names=["HighCPU"],
affected_services=["test-app"],
),
repair_steps=[
RepairStep(
step_number=1,
action_type=ActionType.KUBECTL,
command="kubectl rollout restart deployment/{target}",
risk_level=risk_level,
),
],
success_count=success_count,
failure_count=failure_count,
)
def create_match(
self,
playbook_id: str = "PB-TEST-001",
similarity: float = 0.95,
) -> PlaybookMatch:
"""建立測試用 PlaybookMatch"""
return PlaybookMatch(
playbook_id=playbook_id,
similarity_score=similarity,
match_type="hybrid",
)
# =========================================================================
# 通過條件測試
# =========================================================================
def test_approve_all_conditions_met(self, policy):
"""所有條件都滿足時應該批准"""
proposal = self.create_proposal_data(
risk_level="low",
confidence=0.95,
)
playbook = self.create_playbook(success_count=10, failure_count=0)
match = self.create_match(similarity=0.95)
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is True
assert decision.reason == AutoApproveReason.PLAYBOOK_MATCH
def test_approve_with_high_confidence(self, policy):
"""高信心度應該通過"""
proposal = self.create_proposal_data(confidence=0.99)
playbook = self.create_playbook(success_count=10)
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is True
assert decision.confidence == 0.99
# =========================================================================
# 拒絕條件測試
# =========================================================================
def test_reject_critical_risk(self, policy):
"""CRITICAL 風險永遠拒絕"""
proposal = self.create_proposal_data(risk_level="critical")
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is False
assert decision.reason == AutoApproveReason.CRITICAL_OPERATION
def test_reject_high_risk(self, policy):
"""HIGH 風險應該拒絕"""
proposal = self.create_proposal_data(risk_level="high")
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is False
assert decision.reason == AutoApproveReason.HIGH_RISK
def test_reject_medium_risk(self, policy):
"""MEDIUM 風險應該拒絕 (預設只允許 low)"""
proposal = self.create_proposal_data(risk_level="medium")
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is False
assert decision.reason == AutoApproveReason.HIGH_RISK
def test_reject_low_trust_score(self, policy, mock_trust_manager):
"""低信任分數應該拒絕"""
mock_trust_manager.get_trust_record.return_value = MagicMock(score=2)
proposal = self.create_proposal_data()
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is False
assert decision.reason == AutoApproveReason.LOW_TRUST
def test_reject_low_confidence(self, policy):
"""低信心度應該拒絕"""
proposal = self.create_proposal_data(confidence=0.7)
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is False
assert decision.reason == AutoApproveReason.LOW_TRUST
assert "Confidence" in decision.reason_detail
def test_reject_no_playbook(self, policy):
"""無 Playbook 時應該拒絕"""
proposal = self.create_proposal_data()
decision = policy.evaluate(proposal, None, None)
assert decision.should_auto_approve is False
assert decision.reason == AutoApproveReason.NO_PLAYBOOK
def test_reject_low_playbook_success_rate(self, policy):
"""Playbook 成功率低應該拒絕"""
proposal = self.create_proposal_data()
playbook = self.create_playbook(
success_count=10,
failure_count=5, # 66.7% success rate
)
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is False
assert decision.reason == AutoApproveReason.LOW_SUCCESS_RATE
def test_reject_insufficient_history(self, policy):
"""Playbook 執行次數不足應該拒絕"""
proposal = self.create_proposal_data()
playbook = self.create_playbook(
success_count=2, # < 3
failure_count=0,
)
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is False
assert decision.reason == AutoApproveReason.INSUFFICIENT_HISTORY
# =========================================================================
# 邊界條件測試
# =========================================================================
def test_boundary_trust_score_exactly_5(self, policy, mock_trust_manager):
"""信任分數剛好等於 5 (邊界值)"""
mock_trust_manager.get_trust_record.return_value = MagicMock(score=5)
proposal = self.create_proposal_data()
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is True
assert decision.trust_score == 5
def test_boundary_confidence_exactly_90(self, policy):
"""信心度剛好等於 90% (邊界值)"""
proposal = self.create_proposal_data(confidence=0.90)
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is True
assert decision.confidence == 0.90
def test_boundary_success_count_exactly_3(self, policy):
"""成功次數剛好等於 3 (邊界值)"""
proposal = self.create_proposal_data()
playbook = self.create_playbook(success_count=3, failure_count=0)
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is True
assert decision.playbook_success_count == 3
def test_boundary_success_rate_exactly_95(self, policy):
"""成功率剛好等於 95% (邊界值)"""
proposal = self.create_proposal_data()
playbook = self.create_playbook(
success_count=19,
failure_count=1, # 95% exactly
)
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is True
# =========================================================================
# 配置測試
# =========================================================================
def test_disabled_policy_rejects_all(self, mock_trust_manager):
"""停用的策略應該拒絕所有請求"""
config = AutoApproveConfig(enabled=False)
policy = AutoApprovePolicy(config=config, trust_manager=mock_trust_manager)
proposal = self.create_proposal_data()
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
assert decision.should_auto_approve is False
assert "disabled" in decision.reason_detail.lower()
def test_custom_allowed_risk_levels(self, mock_trust_manager):
"""自訂允許的風險等級"""
config = AutoApproveConfig(allowed_risk_levels=["low", "medium"])
policy = AutoApprovePolicy(config=config, trust_manager=mock_trust_manager)
proposal = self.create_proposal_data(risk_level="medium")
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
# medium 風險現在應該被允許
assert decision.should_auto_approve is True
# =========================================================================
# 資料結構測試
# =========================================================================
def test_decision_to_dict(self, policy):
"""Decision.to_dict() 應該正常工作"""
proposal = self.create_proposal_data()
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
data = decision.to_dict()
assert "should_auto_approve" in data
assert "reason" in data
assert "reason_detail" in data
assert "risk_level" in data
assert "trust_score" in data
assert "confidence" in data
assert "decided_at" in data
def test_decision_to_audit_log(self, policy):
"""Decision.to_audit_log() 應該正常工作"""
proposal = self.create_proposal_data()
playbook = self.create_playbook()
match = self.create_match()
decision = policy.evaluate(proposal, playbook, match)
log = decision.to_audit_log()
assert "AUTO_APPROVED" in log or "REQUIRES_HUMAN" in log
assert "risk=" in log
assert "trust=" in log
class TestAutoApproveReason:
"""AutoApproveReason Enum 測試"""
def test_all_reasons_exist(self):
"""確認所有原因類型都存在"""
# 批准原因
assert AutoApproveReason.PLAYBOOK_MATCH.value == "playbook_match"
assert AutoApproveReason.TRUST_SCORE.value == "trust_score"
assert AutoApproveReason.LOW_RISK.value == "low_risk"
# 拒絕原因
assert AutoApproveReason.HIGH_RISK.value == "high_risk"
assert AutoApproveReason.CRITICAL_OPERATION.value == "critical_operation"
assert AutoApproveReason.LOW_TRUST.value == "low_trust"
assert AutoApproveReason.NO_PLAYBOOK.value == "no_playbook"
assert AutoApproveReason.LOW_SUCCESS_RATE.value == "low_success_rate"
assert AutoApproveReason.INSUFFICIENT_HISTORY.value == "insufficient_history"

326
apps/api/tests/test_adr030_learning_service.py
View File

@@ -1,326 +0,0 @@
"""
ADR-030 Learning Service Tests
==============================
測試持續學習服務
版本: v1.0
建立: 2026-03-26 (台北時區)
建立者: Claude Code (ADR-030 Phase 5)
"""
import uuid
from unittest.mock import MagicMock, patch
import pytest
from src.models.approval import ApprovalRequest, RiskLevel
from src.services.learning_service import (
ExecutionResult,
FeedbackRequest,
FeedbackType,
LearningRecord,
LearningService,
)
class TestExecutionResult:
"""ExecutionResult 資料結構測試"""
def test_create_success_result(self):
"""建立成功執行結果"""
result = ExecutionResult(
approval_id="APR-001",
incident_id="INC-001",
action="kubectl rollout restart",
success=True,
duration_seconds=2.5,
)
assert result.success is True
assert result.error_message is None
assert result.duration_seconds == 2.5
def test_create_failure_result(self):
"""建立失敗執行結果"""
result = ExecutionResult(
approval_id="APR-001",
incident_id="INC-001",
action="kubectl rollout restart",
success=False,
error_message="Pod not found",
)
assert result.success is False
assert result.error_message == "Pod not found"
def test_to_dict(self):
"""to_dict() 應該正常工作"""
result = ExecutionResult(
approval_id="APR-001",
incident_id="INC-001",
action="kubectl rollout restart",
success=True,
duration_seconds=1.5,
)
data = result.to_dict()
assert data["approval_id"] == "APR-001"
assert data["incident_id"] == "INC-001"
assert data["success"] is True
assert "executed_at" in data
class TestFeedbackRequest:
"""FeedbackRequest 資料結構測試"""
def test_create_human_approve_feedback(self):
"""建立人工批准反饋"""
feedback = FeedbackRequest(
incident_id="INC-001",
feedback_type=FeedbackType.HUMAN_APPROVE,
submitted_by="admin",
)
assert feedback.feedback_type == FeedbackType.HUMAN_APPROVE
assert feedback.submitted_by == "admin"
def test_create_effectiveness_rating(self):
"""建立有效性評分反饋"""
feedback = FeedbackRequest(
incident_id="INC-001",
feedback_type=FeedbackType.EFFECTIVENESS_RATING,
effectiveness_score=5,
learning_notes="非常有效的修復",
)
assert feedback.effectiveness_score == 5
assert feedback.learning_notes == "非常有效的修復"
class TestLearningRecord:
"""LearningRecord 資料結構測試"""
def test_create_learning_record(self):
"""建立學習記錄"""
record = LearningRecord(
incident_id="INC-001",
feedback_type=FeedbackType.EXECUTION_SUCCESS,
action_pattern="restart:test-app-*",
trust_before=3,
trust_after=4,
playbook_updated=True,
)
assert record.trust_before == 3
assert record.trust_after == 4
assert record.playbook_updated is True
def test_to_dict(self):
"""to_dict() 應該正常工作"""
record = LearningRecord(
incident_id="INC-001",
feedback_type=FeedbackType.EXECUTION_SUCCESS,
action_pattern="restart:test-app-*",
trust_before=3,
trust_after=4,
)
data = record.to_dict()
assert data["incident_id"] == "INC-001"
assert data["feedback_type"] == "execution_success"
assert data["trust_before"] == 3
assert data["trust_after"] == 4
class TestLearningService:
"""LearningService 單元測試"""
@pytest.fixture
def mock_trust_manager(self):
"""建立 mock TrustScoreManager"""
manager = MagicMock()
manager.get_trust_record.return_value = MagicMock(score=3)
manager.record_approval.return_value = None
manager.record_rejection.return_value = None
return manager
@pytest.fixture
def service(self, mock_trust_manager):
"""建立 LearningService with mocked dependencies"""
with patch("src.services.learning_service.get_trust_manager", return_value=mock_trust_manager):
return LearningService()
def create_approval(self, action: str = "kubectl rollout restart deployment/test-app -n prod") -> ApprovalRequest:
"""建立測試用 ApprovalRequest"""
return ApprovalRequest(
id=uuid.uuid4(),
action=action,
description="Test approval",
risk_level=RiskLevel.LOW,
required_signatures=1,
requested_by="test-system",
)
# =========================================================================
# Action Pattern 提取測試
# =========================================================================
def test_extract_action_pattern_restart(self, service):
"""測試 restart 動作模式提取"""
pattern = service._extract_action_pattern(
"kubectl rollout restart deployment/test-app-abc123-def456 -n prod"
)
# 應該移除 pod hash suffix
assert "restart" in pattern
assert "abc123" not in pattern
def test_extract_action_pattern_delete(self, service):
"""測試 delete 動作模式提取"""
pattern = service._extract_action_pattern(
"kubectl delete pod test-pod-xyz789-abc123 -n staging"
)
assert "delete" in pattern
assert "xyz789" not in pattern
def test_extract_action_pattern_empty(self, service):
"""測試空動作"""
pattern = service._extract_action_pattern("")
assert pattern == "unknown"
def test_extract_action_pattern_short(self, service):
"""測試太短的動作"""
pattern = service._extract_action_pattern("kubectl")
assert pattern == "unknown"
# =========================================================================
# 執行結果處理測試
# =========================================================================
@pytest.mark.asyncio
async def test_process_success_result(self, service, mock_trust_manager):
"""處理成功執行結果"""
approval = self.create_approval()
result = ExecutionResult(
approval_id="apr-001",
incident_id="INC-001",
action=approval.action,
success=True,
duration_seconds=2.0,
)
record = await service.process_execution_result(approval, result)
assert record.feedback_type == FeedbackType.EXECUTION_SUCCESS
mock_trust_manager.record_approval.assert_called_once()
@pytest.mark.asyncio
async def test_process_failure_result(self, service, mock_trust_manager):
"""處理失敗執行結果"""
approval = self.create_approval()
result = ExecutionResult(
approval_id="apr-001",
incident_id="INC-001",
action=approval.action,
success=False,
error_message="Pod not found",
)
record = await service.process_execution_result(approval, result)
assert record.feedback_type == FeedbackType.EXECUTION_FAILURE
mock_trust_manager.record_rejection.assert_called_once()
# =========================================================================
# 人工反饋處理測試
# =========================================================================
@pytest.mark.asyncio
async def test_process_human_approve(self, service, mock_trust_manager):
"""處理人工批准反饋"""
feedback = FeedbackRequest(
incident_id="INC-001",
feedback_type=FeedbackType.HUMAN_APPROVE,
submitted_by="admin",
)
record = await service.process_human_feedback(feedback)
assert record.feedback_type == FeedbackType.HUMAN_APPROVE
mock_trust_manager.record_approval.assert_called_once()
@pytest.mark.asyncio
async def test_process_human_reject(self, service, mock_trust_manager):
"""處理人工拒絕反饋"""
feedback = FeedbackRequest(
incident_id="INC-001",
feedback_type=FeedbackType.HUMAN_REJECT,
submitted_by="admin",
)
record = await service.process_human_feedback(feedback)
assert record.feedback_type == FeedbackType.HUMAN_REJECT
mock_trust_manager.record_rejection.assert_called_once()
@pytest.mark.asyncio
async def test_process_high_effectiveness_rating(self, service, mock_trust_manager):
"""處理高有效性評分 (4-5 分)"""
feedback = FeedbackRequest(
incident_id="INC-001",
feedback_type=FeedbackType.EFFECTIVENESS_RATING,
effectiveness_score=5,
)
record = await service.process_human_feedback(feedback)
assert record.feedback_type == FeedbackType.EFFECTIVENESS_RATING
mock_trust_manager.record_approval.assert_called_once()
@pytest.mark.asyncio
async def test_process_low_effectiveness_rating(self, service, mock_trust_manager):
"""處理低有效性評分 (1-2 分)"""
feedback = FeedbackRequest(
incident_id="INC-001",
feedback_type=FeedbackType.EFFECTIVENESS_RATING,
effectiveness_score=1,
)
record = await service.process_human_feedback(feedback)
assert record.feedback_type == FeedbackType.EFFECTIVENESS_RATING
mock_trust_manager.record_rejection.assert_called_once()
@pytest.mark.asyncio
async def test_process_medium_effectiveness_rating(self, service, mock_trust_manager):
"""處理中等有效性評分 (3 分) - 不調整信任度"""
feedback = FeedbackRequest(
incident_id="INC-001",
feedback_type=FeedbackType.EFFECTIVENESS_RATING,
effectiveness_score=3,
)
record = await service.process_human_feedback(feedback)
assert record.feedback_type == FeedbackType.EFFECTIVENESS_RATING
# 中等評分不應該調整信任度
mock_trust_manager.record_approval.assert_not_called()
mock_trust_manager.record_rejection.assert_not_called()
class TestFeedbackType:
"""FeedbackType Enum 測試"""
def test_all_feedback_types_exist(self):
"""確認所有反饋類型都存在"""
assert FeedbackType.EXECUTION_SUCCESS.value == "execution_success"
assert FeedbackType.EXECUTION_FAILURE.value == "execution_failure"
assert FeedbackType.HUMAN_APPROVE.value == "human_approve"
assert FeedbackType.HUMAN_REJECT.value == "human_reject"
assert FeedbackType.HUMAN_OVERRIDE.value == "human_override"
assert FeedbackType.EFFECTIVENESS_RATING.value == "effectiveness_rating"

196
apps/api/tests/test_anomaly_counter.py Normal file
View File

@@ -0,0 +1,196 @@
"""
AnomalyCounter 整合測試
=======================
ADR-037: 監控增強架構 - 異常頻率統計
🔴🔴 遵循「禁止 Mock 測試鐵律」- 使用真實 Redis
建立: 2026-03-29 (台北時區) Claude Code
"""
import pytest
from src.services.anomaly_counter import AnomalyCounter, AnomalyFrequency
class TestAnomalyCounterHashSignature:
"""測試異常簽名 Hash 生成"""
def test_same_input_same_hash(self):
"""相同輸入應產生相同 hash"""
sig1 = {"alert_name": "PodCrash", "service": "api"}
sig2 = {"alert_name": "PodCrash", "service": "api"}
assert AnomalyCounter.hash_signature(sig1) == AnomalyCounter.hash_signature(sig2)
def test_different_input_different_hash(self):
"""不同輸入應產生不同 hash"""
sig1 = {"alert_name": "PodCrash", "service": "api"}
sig2 = {"alert_name": "PodCrash", "service": "web"}
assert AnomalyCounter.hash_signature(sig1) != AnomalyCounter.hash_signature(sig2)
def test_ignores_extra_fields(self):
"""應忽略非關鍵欄位 (如 timestamp)"""
sig1 = {"alert_name": "PodCrash", "service": "api"}
sig2 = {"alert_name": "PodCrash", "service": "api", "timestamp": "2026-01-01"}
assert AnomalyCounter.hash_signature(sig1) == AnomalyCounter.hash_signature(sig2)
def test_alertname_alias(self):
"""應支援 alertname (Prometheus 格式) 別名"""
sig1 = {"alert_name": "PodCrash", "service": "api"}
sig2 = {"alertname": "PodCrash", "service": "api"}
assert AnomalyCounter.hash_signature(sig1) == AnomalyCounter.hash_signature(sig2)
def test_job_alias(self):
"""應支援 job (Prometheus 格式) 別名"""
sig1 = {"alert_name": "PodCrash", "service": "api"}
sig2 = {"alert_name": "PodCrash", "job": "api"}
assert AnomalyCounter.hash_signature(sig1) == AnomalyCounter.hash_signature(sig2)
class TestAnomalyFrequencyToDict:
"""測試 AnomalyFrequency.to_dict()"""
def test_to_dict_returns_all_fields(self):
"""to_dict 應返回所有欄位"""
from datetime import datetime
freq = AnomalyFrequency(
anomaly_key="abc123",
count_1h=3,
count_24h=10,
count_7d=50,
count_30d=200,
first_seen=datetime(2026, 3, 1),
last_seen=datetime(2026, 3, 29),
auto_repair_count=5,
permanent_fix_applied=False,
escalation_level="ESCALATE",
)
d = freq.to_dict()
assert d["anomaly_key"] == "abc123"
assert d["count_1h"] == 3
assert d["count_24h"] == 10
assert d["count_7d"] == 50
assert d["count_30d"] == 200
assert d["auto_repair_count"] == 5
assert d["permanent_fix_applied"] is False
assert d["escalation_level"] == "ESCALATE"
class TestAnomalyCounterEscalationLevel:
"""測試升級等級判斷"""
def test_no_escalation_below_threshold(self):
"""低於閾值不應升級"""
counter = AnomalyCounter.__new__(AnomalyCounter)
assert counter._get_escalation_level(2) is None
def test_repeat_at_threshold(self):
"""達到 REPEAT 閾值 (3 次)"""
counter = AnomalyCounter.__new__(AnomalyCounter)
assert counter._get_escalation_level(3) == "REPEAT"
def test_escalate_at_threshold(self):
"""達到 ESCALATE 閾值 (5 次)"""
counter = AnomalyCounter.__new__(AnomalyCounter)
assert counter._get_escalation_level(5) == "ESCALATE"
def test_permanent_fix_at_threshold(self):
"""達到 PERMANENT_FIX 閾值 (10 次)"""
counter = AnomalyCounter.__new__(AnomalyCounter)
assert counter._get_escalation_level(10) == "PERMANENT_FIX"
def test_highest_level_wins(self):
"""超過最高閾值應返回最高等級"""
counter = AnomalyCounter.__new__(AnomalyCounter)
assert counter._get_escalation_level(100) == "PERMANENT_FIX"
# =============================================================================
# 整合測試 (需要真實 Redis)
# =============================================================================
@pytest.mark.asyncio
@pytest.mark.integration
class TestAnomalyCounterIntegration:
"""
整合測試 - 需要真實 Redis
執行方式: pytest -m integration tests/test_anomaly_counter.py
"""
@pytest.fixture
async def counter(self):
"""取得 AnomalyCounter 實例"""
from src.core.redis_client import get_redis
redis = get_redis()
counter = AnomalyCounter(redis)
# 清理測試資料
test_key = "test_anomaly_key"
await redis.delete(f"anomaly:timeline:{test_key}")
await redis.delete(f"anomaly:repair_count:{test_key}")
await redis.delete(f"anomaly:permanent_fix:{test_key}")
await redis.delete(f"anomaly:metadata:{test_key}")
await redis.delete(f"anomaly:repair_history:{test_key}")
return counter
async def test_record_anomaly_returns_frequency(self, counter):
"""record_anomaly 應返回頻率統計"""
signature = {
"alert_name": "TestAlert",
"service": "test-service",
"namespace": "test",
}
freq = await counter.record_anomaly(signature)
assert freq.count_1h >= 1
assert freq.count_24h >= 1
assert freq.anomaly_key is not None
async def test_record_anomaly_increments_count(self, counter):
"""多次記錄應遞增計數"""
signature = {
"alert_name": "TestAlert",
"service": "test-service",
"namespace": "test",
}
freq1 = await counter.record_anomaly(signature)
_ = await counter.record_anomaly(signature)
freq3 = await counter.record_anomaly(signature)
assert freq3.count_1h == freq1.count_1h + 2
assert freq3.count_24h == freq1.count_24h + 2
async def test_record_repair_attempt(self, counter):
"""記錄修復嘗試"""
anomaly_key = "test_repair_key"
await counter.record_repair_attempt(anomaly_key, "restart_pod", True)
await counter.record_repair_attempt(anomaly_key, "restart_pod", False)
stats = await counter.get_repair_success_rate(anomaly_key, "restart_pod")
assert stats["total"] == 2
assert stats["success"] == 1
assert stats["success_rate"] == 0.5
async def test_should_skip_action_low_success_rate(self, counter):
"""低成功率動作應跳過"""
anomaly_key = "test_skip_key"
# 模擬多次失敗
await counter.record_repair_attempt(anomaly_key, "bad_action", False)
await counter.record_repair_attempt(anomaly_key, "bad_action", False)
await counter.record_repair_attempt(anomaly_key, "bad_action", False)
should_skip = await counter.should_skip_action(anomaly_key, "bad_action")
assert should_skip is True

3
apps/api/tests/test_approval_field_alignment.py
View File

@@ -18,14 +18,11 @@ from pydantic import ValidationError
from src.models.approval import (
ApprovalRequestCreate,
ApprovalRequestBase,
BlastRadius,
DataImpact,
DryRunCheck,
RiskLevel,
)
# =============================================================================
# Test: ApprovalRequestCreate Schema 驗證
# =============================================================================

11
apps/api/tests/test_nvidia_provider.py
View File

@@ -6,7 +6,6 @@ NVIDIA Provider Tests - ADR-036
注意: 這些是單元測試，不需要真實的 NVIDIA API Key
"""
import json
import pytest
@@ -407,7 +406,7 @@ class TestAIRouterNvidiaIntegration:
def test_tool_calling_route(self):
"""測試 Tool Calling 路由"""
from src.services.ai_router import get_ai_router, AIProvider, reset_ai_router
from src.services.ai_router import AIProvider, get_ai_router, reset_ai_router
reset_ai_router()
router = get_ai_router()
@@ -425,7 +424,7 @@ class TestAIRouterNvidiaIntegration:
def test_existing_routing_not_affected(self):
"""測試現有路由規則不受影響"""
from src.services.ai_router import get_ai_router, AIProvider, reset_ai_router
from src.services.ai_router import AIProvider, get_ai_router, reset_ai_router
reset_ai_router()
router = get_ai_router()
@@ -568,9 +567,9 @@ class TestPrometheusMetrics:
def test_metrics_defined(self):
"""測試 Prometheus Metrics 已定義"""
from src.services.nvidia_provider import (
NVIDIA_REQUESTS_TOTAL,
NVIDIA_LATENCY_HISTOGRAM,
NVIDIA_CIRCUIT_BREAKER_STATE,
NVIDIA_LATENCY_HISTOGRAM,
NVIDIA_REQUESTS_TOTAL,
)
assert NVIDIA_REQUESTS_TOTAL is not None
@@ -593,8 +592,8 @@ class TestExponentialBackoff:
"""測試退避常數已定義"""
from src.services.nvidia_provider import (
RETRY_BASE_DELAY,
RETRY_MAX_DELAY,
RETRY_EXPONENTIAL_BASE,
RETRY_MAX_DELAY,
)
assert RETRY_BASE_DELAY == 1.0

3
apps/api/tests/test_telegram_integration.py
View File

@@ -13,12 +13,11 @@ P1-4: Telegram 整合驗證測試
建立者: Claude Code (P1-4 Telegram 驗證)
"""
import pytest
from src.services.telegram_gateway import (
RISK_EMOJI_MAP,
SignOzMetricsBlock,
TelegramMessage,
RISK_EMOJI_MAP,
)
# 本地定義 (避免 import 錯誤)