684d6cfb43
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 17m34s
Task 2: AlertGroupingService — Redis 5分鐘滑動視窗，防告警風暴 - apps/api/src/services/alert_grouping_service.py (新增) - webhooks.py 整合：指紋生成後/LLM前短路子告警 - Threshold=3，Graceful Degradation，16 tests Task 3: approval_execution.py 執行失敗重試 - MAX_RETRY=2, RETRY_DELAY_SECONDS=30 - _is_transient_error() 瞬態/永久分類，永久錯誤不重試 - Timeline 記錄重試進度，成功後標注重試次數，29 tests Task 4: report_generation_service.py 自動報告 - 日度巡檢報告：每日 08:00 台北時間，Telegram SRE 群組推送 - Postmortem：Incident resolved + duration > 10 分鐘自動觸發 - main.py lifespan 掛載 run_daily_report_loop()，30 tests 測試: 600 → 675 通過 (+75)，0 failed Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
135 lines
5.1 KiB
Python
135 lines
5.1 KiB
Python
"""
|
||
ApprovalExecutionService 重試邏輯單元測試
|
||
==========================================
|
||
ADR-076 Task 3: 執行失敗重試機制
|
||
|
||
測試範圍:
|
||
- _is_transient_error() 瞬態/永久性錯誤分類
|
||
- MAX_RETRY / RETRY_DELAY_SECONDS 常數
|
||
- 邊界情境: None、空字串、混合訊息
|
||
|
||
🔴🔴 遵循「禁止 Mock 測試鐵律」
|
||
- _is_transient_error 是純 Python 方法，無 DB/Redis 依賴
|
||
- 無需 Mock，直接測試真實邏輯
|
||
|
||
建立: 2026-04-14 (台北時區) Claude Haiku 4.5
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from src.services.approval_execution import ApprovalExecutionService
|
||
|
||
|
||
class TestIsTransientError:
|
||
"""測試瞬態/永久性錯誤判斷邏輯"""
|
||
|
||
# ------- 瞬態錯誤（應返回 True）-------
|
||
|
||
def test_connection_refused(self):
|
||
assert ApprovalExecutionService._is_transient_error("connection refused") is True
|
||
|
||
def test_connection_refused_uppercase(self):
|
||
"""大小寫不敏感"""
|
||
assert ApprovalExecutionService._is_transient_error("Connection Refused") is True
|
||
|
||
def test_timeout(self):
|
||
assert ApprovalExecutionService._is_transient_error("request timeout") is True
|
||
|
||
def test_timed_out(self):
|
||
assert ApprovalExecutionService._is_transient_error("operation timed out") is True
|
||
|
||
def test_io_error(self):
|
||
assert ApprovalExecutionService._is_transient_error("i/o error reading response") is True
|
||
|
||
def test_io_error_alt(self):
|
||
assert ApprovalExecutionService._is_transient_error("io error") is True
|
||
|
||
def test_service_unavailable(self):
|
||
assert ApprovalExecutionService._is_transient_error("service unavailable") is True
|
||
|
||
def test_too_many_requests(self):
|
||
assert ApprovalExecutionService._is_transient_error("too many requests") is True
|
||
|
||
def test_eof(self):
|
||
assert ApprovalExecutionService._is_transient_error("unexpected eof") is True
|
||
|
||
def test_dial_tcp(self):
|
||
assert ApprovalExecutionService._is_transient_error("dial tcp 10.0.0.1:6443: connect") is True
|
||
|
||
def test_connection_reset(self):
|
||
assert ApprovalExecutionService._is_transient_error("connection reset by peer") is True
|
||
|
||
def test_temporary_failure(self):
|
||
assert ApprovalExecutionService._is_transient_error("temporary failure in name resolution") is True
|
||
|
||
# ------- 永久性錯誤（應返回 False）-------
|
||
|
||
def test_not_found(self):
|
||
assert ApprovalExecutionService._is_transient_error("pod not found") is False
|
||
|
||
def test_forbidden(self):
|
||
assert ApprovalExecutionService._is_transient_error("forbidden: insufficient permissions") is False
|
||
|
||
def test_permission_denied(self):
|
||
assert ApprovalExecutionService._is_transient_error("permission denied") is False
|
||
|
||
def test_unauthorized(self):
|
||
assert ApprovalExecutionService._is_transient_error("unauthorized") is False
|
||
|
||
def test_already_exists(self):
|
||
assert ApprovalExecutionService._is_transient_error("resource already exists") is False
|
||
|
||
def test_invalid(self):
|
||
assert ApprovalExecutionService._is_transient_error("invalid field selector") is False
|
||
|
||
def test_destructive_blocked(self):
|
||
assert ApprovalExecutionService._is_transient_error("destructive operation blocked") is False
|
||
|
||
def test_immutable(self):
|
||
assert ApprovalExecutionService._is_transient_error("field is immutable") is False
|
||
|
||
# ------- 邊界情境 -------
|
||
|
||
def test_none_returns_false(self):
|
||
"""None → 不重試（無法判斷）"""
|
||
assert ApprovalExecutionService._is_transient_error(None) is False
|
||
|
||
def test_empty_string_returns_false(self):
|
||
"""空字串 → 不重試"""
|
||
assert ApprovalExecutionService._is_transient_error("") is False
|
||
|
||
def test_permanent_wins_over_transient(self):
|
||
"""混合訊息：永久性錯誤關鍵字優先，不重試"""
|
||
# "not found" (永久) + "timeout" (瞬態) → 不重試
|
||
assert ApprovalExecutionService._is_transient_error("timeout: pod not found") is False
|
||
|
||
def test_unknown_error_not_retried(self):
|
||
"""未知錯誤不重試"""
|
||
assert ApprovalExecutionService._is_transient_error("kubectl exited with code 1") is False
|
||
|
||
|
||
class TestRetryConstants:
|
||
"""測試重試常數設定"""
|
||
|
||
def test_max_retry(self):
|
||
"""最多重試 2 次（共 3 次嘗試）"""
|
||
assert ApprovalExecutionService.MAX_RETRY == 2
|
||
|
||
def test_retry_delay(self):
|
||
"""重試間隔 30 秒"""
|
||
assert ApprovalExecutionService.RETRY_DELAY_SECONDS == 30
|
||
|
||
def test_transient_keywords_not_empty(self):
|
||
"""瞬態錯誤關鍵字列表不為空"""
|
||
assert len(ApprovalExecutionService._TRANSIENT_ERROR_KEYWORDS) > 0
|
||
|
||
def test_permanent_keywords_not_empty(self):
|
||
"""永久性錯誤關鍵字列表不為空"""
|
||
assert len(ApprovalExecutionService._PERMANENT_ERROR_KEYWORDS) > 0
|
||
|
||
def test_no_overlap_in_keywords(self):
|
||
"""瞬態/永久性關鍵字不重疊（避免邏輯衝突）"""
|
||
transient = set(ApprovalExecutionService._TRANSIENT_ERROR_KEYWORDS)
|
||
permanent = set(ApprovalExecutionService._PERMANENT_ERROR_KEYWORDS)
|
||
assert transient.isdisjoint(permanent)
|