59c9eff83a
- F401: 移除未使用的 imports (TerminalSessionStatus, AutoApproveDecision, TerminalSession)
- I001: 修正 import blocks 排序
- C401: set(generator) → {set comprehension}
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
327 lines
11 KiB
Python
327 lines
11 KiB
Python
"""
|
|
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"
|