242 lines
10 KiB
Python
242 lines
10 KiB
Python
"""
|
|
단위 테스트: ChatOps 확장 + 예측 장애 방지
|
|
|
|
커버 항목:
|
|
- chatops_extended 라우터 임포트 및 prefix 검증
|
|
- predictive_failure 라우터 임포트 및 prefix 검증
|
|
- ORM 모델 테이블명 및 컬럼 검증 (ChatOpsCommand, FailureSignal, PreventionAction)
|
|
- 명령어 파서 (_parse_command) 단위 검증
|
|
- 리스크 점수 계산 (_calc_risk_score) 단위 검증
|
|
- 예측 장애 레이블 (_predict_failure_label) 검증
|
|
- 지원 채널 정의 일관성 검증
|
|
- 장애 패턴 모델 정의 검증
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# chatops_extended 라우터 검증
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def test_chatops_extended_import():
|
|
"""chatops_extended 모듈이 오류 없이 임포트된다."""
|
|
from routers import chatops_extended
|
|
assert chatops_extended.router is not None
|
|
|
|
|
|
def test_chatops_router_prefix():
|
|
"""라우터 prefix가 /api/chatops인지 확인."""
|
|
from routers.chatops_extended import router
|
|
assert router.prefix == "/api/chatops"
|
|
|
|
|
|
def test_chatops_supported_channels():
|
|
"""지원 채널 3종(kakao, slack, internal)이 정의되어 있다."""
|
|
from routers.chatops_extended import SUPPORTED_CHANNELS
|
|
assert "kakao" in SUPPORTED_CHANNELS
|
|
assert "slack" in SUPPORTED_CHANNELS
|
|
assert "internal" in SUPPORTED_CHANNELS
|
|
# 각 채널에 enabled 키가 있어야 한다
|
|
for ch_id, info in SUPPORTED_CHANNELS.items():
|
|
assert "enabled" in info
|
|
assert "name" in info
|
|
|
|
|
|
def test_chatops_command_definitions():
|
|
"""지원 명령어가 7개 이상 정의되어 있다."""
|
|
from routers.chatops_extended import COMMAND_DEFINITIONS
|
|
commands = [d["command"] for d in COMMAND_DEFINITIONS]
|
|
assert len(commands) >= 7
|
|
# 필수 명령어 포함 여부
|
|
assert "/sr create" in commands
|
|
assert "/status" in commands
|
|
assert "/deploy" in commands
|
|
assert "/approve" in commands
|
|
assert "/report" in commands
|
|
assert "/patch" in commands
|
|
assert "/workflow" in commands
|
|
|
|
|
|
def test_parse_command_slash_sr_create():
|
|
"""'/sr create 서버 재시작 본문' 파싱 결과 확인."""
|
|
from routers.chatops_extended import _parse_command
|
|
result = _parse_command("/sr create web-01 재시작 요청")
|
|
assert result is not None
|
|
assert result["command"] == "/sr create"
|
|
assert "web-01" in result["args"]
|
|
|
|
|
|
def test_parse_command_status_with_id():
|
|
"""'/status SR-2026-001' 파싱 결과 확인."""
|
|
from routers.chatops_extended import _parse_command
|
|
result = _parse_command("/status SR-2026-001")
|
|
assert result is not None
|
|
assert result["command"] == "/status"
|
|
|
|
|
|
def test_parse_command_unknown_returns_dict():
|
|
"""인식되지 않는 명령어도 dict를 반환한다 (None 반환 없음)."""
|
|
from routers.chatops_extended import _parse_command
|
|
result = _parse_command("/unknown_cmd arg1 arg2")
|
|
assert result is not None
|
|
assert "command" in result
|
|
|
|
|
|
def test_parse_command_no_slash_returns_none():
|
|
"""슬래시 없는 일반 메시지는 None을 반환한다."""
|
|
from routers.chatops_extended import _parse_command
|
|
result = _parse_command("안녕하세요 도움이 필요합니다")
|
|
assert result is None
|
|
|
|
|
|
def test_parse_command_empty_string_returns_none():
|
|
"""빈 문자열은 None을 반환한다."""
|
|
from routers.chatops_extended import _parse_command
|
|
result = _parse_command("")
|
|
assert result is None
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# predictive_failure 라우터 검증
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def test_predictive_failure_import():
|
|
"""predictive_failure 모듈이 오류 없이 임포트된다."""
|
|
from routers import predictive_failure
|
|
assert predictive_failure.router is not None
|
|
|
|
|
|
def test_predictive_failure_router_prefix():
|
|
"""라우터 prefix가 /api/predict-fail인지 확인."""
|
|
from routers.predictive_failure import router
|
|
assert router.prefix == "/api/predict-fail"
|
|
|
|
|
|
def test_failure_patterns_defined():
|
|
"""장애 패턴 모델이 4종 이상 정의되어 있다."""
|
|
from routers.predictive_failure import FAILURE_PATTERNS
|
|
assert len(FAILURE_PATTERNS) >= 4
|
|
signal_types = {p["signal_type"] for p in FAILURE_PATTERNS}
|
|
assert "cpu_spike" in signal_types
|
|
assert "mem_leak" in signal_types
|
|
assert "disk_full" in signal_types
|
|
assert "error_rate" in signal_types
|
|
|
|
|
|
def test_failure_pattern_schema():
|
|
"""각 패턴 모델에 필수 키가 존재한다."""
|
|
from routers.predictive_failure import FAILURE_PATTERNS
|
|
required_keys = {"id", "signal_type", "name", "description", "threshold", "window_days", "algorithm"}
|
|
for p in FAILURE_PATTERNS:
|
|
for key in required_keys:
|
|
assert key in p, f"패턴 '{p.get('id', '?')}'에 '{key}' 키 누락"
|
|
|
|
|
|
def test_prevention_templates_coverage():
|
|
"""예방 조치 템플릿이 4종 신호 유형을 모두 커버한다."""
|
|
from routers.predictive_failure import PREVENTION_TEMPLATES
|
|
for sig_type in ("cpu_spike", "mem_leak", "disk_full", "error_rate"):
|
|
assert sig_type in PREVENTION_TEMPLATES
|
|
tpl = PREVENTION_TEMPLATES[sig_type]
|
|
assert "action_type" in tpl
|
|
assert "action_cmd" in tpl
|
|
assert "description" in tpl
|
|
|
|
|
|
def test_calc_risk_score_below_threshold():
|
|
"""임계값 미만 값에서 리스크 점수가 0.8 이하이다."""
|
|
from routers.predictive_failure import _calc_risk_score
|
|
score = _calc_risk_score(70.0, 85.0, "cpu_spike")
|
|
assert 0.0 <= score <= 1.0
|
|
# 임계값 미만이므로 1.0 미만이어야 함
|
|
assert score < 1.0
|
|
|
|
|
|
def test_calc_risk_score_above_threshold():
|
|
"""임계값을 초과하면 리스크 점수가 높다 (0.5 초과)."""
|
|
from routers.predictive_failure import _calc_risk_score
|
|
score = _calc_risk_score(95.0, 85.0, "cpu_spike")
|
|
assert score > 0.5
|
|
|
|
|
|
def test_calc_risk_score_disk_full_high_weight():
|
|
"""disk_full 신호는 가중치 1.0이므로 다른 타입 대비 높다."""
|
|
from routers.predictive_failure import _calc_risk_score
|
|
disk_score = _calc_risk_score(95.0, 95.0, "disk_full")
|
|
cpu_score = _calc_risk_score(95.0, 95.0, "cpu_spike")
|
|
# disk_full(1.0) >= cpu_spike(0.8)
|
|
assert disk_score >= cpu_score
|
|
|
|
|
|
def test_calc_risk_score_zero_threshold():
|
|
"""임계값이 0이면 리스크 점수 0.0 반환 (ZeroDivision 없음)."""
|
|
from routers.predictive_failure import _calc_risk_score
|
|
score = _calc_risk_score(50.0, 0.0, "cpu_spike")
|
|
assert score == 0.0
|
|
|
|
|
|
def test_predict_failure_label_low_risk():
|
|
"""리스크 점수 0.3 미만은 None 반환 (장애 예측 없음)."""
|
|
from routers.predictive_failure import _predict_failure_label
|
|
label = _predict_failure_label("cpu_spike", 0.3)
|
|
assert label is None
|
|
|
|
|
|
def test_predict_failure_label_high_risk():
|
|
"""리스크 점수 0.7 이상은 레이블 문자열 반환."""
|
|
from routers.predictive_failure import _predict_failure_label
|
|
label = _predict_failure_label("mem_leak", 0.8)
|
|
assert isinstance(label, str)
|
|
assert len(label) > 0
|
|
|
|
|
|
def test_predict_failure_label_disk_full():
|
|
"""disk_full 신호의 레이블에 '디스크' 또는 '쓰기' 포함."""
|
|
from routers.predictive_failure import _predict_failure_label
|
|
label = _predict_failure_label("disk_full", 0.9)
|
|
assert label is not None
|
|
assert any(kw in label for kw in ("디스크", "쓰기", "Disk", "Full"))
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# ORM 모델 검증
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def test_chatops_command_orm():
|
|
"""ChatOpsCommand ORM 모델의 테이블명과 컬럼을 확인한다."""
|
|
from models import ChatOpsCommand
|
|
assert ChatOpsCommand.__tablename__ == "tb_chatops_command"
|
|
cols = {c.name for c in ChatOpsCommand.__table__.columns}
|
|
for col in ("id", "channel", "command", "args", "user_id", "response", "success", "created_at"):
|
|
assert col in cols, f"ChatOpsCommand에 '{col}' 컬럼 누락"
|
|
|
|
|
|
def test_failure_signal_orm():
|
|
"""FailureSignal ORM 모델의 테이블명과 컬럼을 확인한다."""
|
|
from models import FailureSignal
|
|
assert FailureSignal.__tablename__ == "tb_failure_signal"
|
|
cols = {c.name for c in FailureSignal.__table__.columns}
|
|
for col in ("id", "server_name", "signal_type", "value", "threshold", "risk_score",
|
|
"predicted_failure", "created_at"):
|
|
assert col in cols, f"FailureSignal에 '{col}' 컬럼 누락"
|
|
|
|
|
|
def test_prevention_action_orm():
|
|
"""PreventionAction ORM 모델의 테이블명과 컬럼을 확인한다."""
|
|
from models import PreventionAction
|
|
assert PreventionAction.__tablename__ == "tb_prevention_action"
|
|
cols = {c.name for c in PreventionAction.__table__.columns}
|
|
for col in ("id", "signal_id", "action_type", "action_cmd", "success", "created_at"):
|
|
assert col in cols, f"PreventionAction에 '{col}' 컬럼 누락"
|
|
|
|
|
|
def test_prevention_action_fk_signal():
|
|
"""PreventionAction.signal_id가 tb_failure_signal을 참조한다."""
|
|
from models import PreventionAction
|
|
fk_targets = {
|
|
str(fk.column) for fk in PreventionAction.__table__.foreign_keys
|
|
}
|
|
assert any("tb_failure_signal" in t for t in fk_targets)
|