cfe2901a55
- itsm/ -> workspace/guardia-itsm/ - manager/ -> workspace/guardia-manager/ - app/ -> workspace/guardia-messenger/ - manual/ -> workspace/guardia-docs/ workspace/zioinfo-web/ unchanged. git mv preserves full commit history. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""B-6 예측 유지보수 테스트"""
|
|
import sys, ast, os, math
|
|
|
|
os.environ.setdefault("GUARDIA_SECRET_KEY", "test-b6-secret-key-32bytes-padded!")
|
|
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_b6.db")
|
|
|
|
ok = True
|
|
|
|
print("=== 1. 구문 검사 ===")
|
|
files = ["core/predictive.py", "routers/predictive.py", "main.py"]
|
|
for f in files:
|
|
try:
|
|
with open(f, encoding="utf-8") as fh:
|
|
src = fh.read()
|
|
ast.parse(src)
|
|
print(f" OK {f}")
|
|
except SyntaxError as e:
|
|
print(f" ERR {f}: {e}")
|
|
ok = False
|
|
|
|
print("\n=== 2. core/predictive.py 함수 확인 ===")
|
|
with open("core/predictive.py", encoding="utf-8") as f:
|
|
pred_src = f.read()
|
|
|
|
fn_checks = [
|
|
("def linear_regression(", "선형 회귀 함수"),
|
|
("def predict_value(", "값 예측 함수"),
|
|
("def time_to_reach(", "임계값 도달 시간 계산"),
|
|
("def moving_average(", "이동 평균"),
|
|
("def detect_seasonal_pattern(", "계절성 패턴 감지"),
|
|
("async def fetch_metric_history(", "메트릭 이력 조회"),
|
|
("async def predict_metric_trend(", "메트릭 트렌드 예측"),
|
|
("async def analyze_server_health(", "서버 건강도 분석"),
|
|
("async def create_preventive_sr(", "예방 SR 자동 생성"),
|
|
("def assess_equipment_lifecycle(", "장비 수명 주기 평가"),
|
|
("async def run_lifecycle_analysis(", "수명 주기 배치 분석"),
|
|
("async def run_predictive_batch(", "예측 배치 실행"),
|
|
("PREDICTION_THRESHOLDS", "예측 임계값 설정"),
|
|
("EQUIPMENT_LIFESPAN", "장비 수명 기준"),
|
|
("TTR", "TTR 관련 로직 (time-to-reach)"),
|
|
("r_squared", "R² 결정계수"),
|
|
]
|
|
for sym, desc in fn_checks:
|
|
status = "OK" if sym in pred_src else "ERR"
|
|
if status == "ERR":
|
|
ok = False
|
|
print(f" {status} {desc}")
|
|
|
|
print("\n=== 3. routers/predictive.py 엔드포인트 확인 ===")
|
|
with open("routers/predictive.py", encoding="utf-8") as f:
|
|
router_src = f.read()
|
|
|
|
endpoint_checks = [
|
|
('@router.post("/analyze/{source}"', "POST /analyze/{source}"),
|
|
('@router.get("/health/{source}"', "GET /health/{source}"),
|
|
('@router.post("/batch"', "POST /batch"),
|
|
('@router.get("/lifecycle"', "GET /lifecycle"),
|
|
('@router.get("/lifecycle/{source}"', "GET /lifecycle/{source}"),
|
|
('@router.get("/thresholds"', "GET /thresholds"),
|
|
('@router.put("/thresholds/{metric_type}"', "PUT /thresholds/{metric}"),
|
|
('@router.get("/stats"', "GET /stats"),
|
|
('@router.post("/simulate"', "POST /simulate"),
|
|
]
|
|
for sym, desc in endpoint_checks:
|
|
status = "OK" if sym in router_src else "ERR"
|
|
if status == "ERR":
|
|
ok = False
|
|
print(f" {status} {desc}")
|
|
|
|
print("\n=== 4. main.py 등록 확인 ===")
|
|
with open("main.py", encoding="utf-8") as f:
|
|
main_src = f.read()
|
|
for sym, desc in [("predictive", "predictive 임포트"), ("predictive.router", "predictive 라우터 등록")]:
|
|
status = "OK" if sym in main_src else "ERR"
|
|
if status == "ERR":
|
|
ok = False
|
|
print(f" {status} {desc}")
|
|
|
|
print("\n=== 5. linear_regression 수학 검증 ===")
|
|
try:
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("pred_mod", "core/predictive.py")
|
|
pred_mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(pred_mod)
|
|
|
|
# 완벽한 선형 데이터 (slope=2, intercept=1)
|
|
x = [0.0, 1.0, 2.0, 3.0, 4.0]
|
|
y = [1.0, 3.0, 5.0, 7.0, 9.0]
|
|
slope, intercept, r_sq = pred_mod.linear_regression(x, y)
|
|
assert abs(slope - 2.0) < 0.001, f"slope 오류: {slope}"
|
|
assert abs(intercept - 1.0) < 0.001, f"intercept 오류: {intercept}"
|
|
assert abs(r_sq - 1.0) < 0.001, f"R² 오류: {r_sq}"
|
|
print(f" OK 완벽한 선형: slope={slope:.3f}, intercept={intercept:.3f}, R²={r_sq:.4f}")
|
|
|
|
# 노이즈가 있는 데이터
|
|
import random; random.seed(42)
|
|
x2 = [float(i) for i in range(50)]
|
|
y2 = [2.0 * i + 10 + random.gauss(0, 1) for i in range(50)]
|
|
slope2, intercept2, r_sq2 = pred_mod.linear_regression(x2, y2)
|
|
assert 1.8 < slope2 < 2.2, f"노이즈 slope 범위 오류: {slope2}"
|
|
assert r_sq2 > 0.95, f"R² 너무 낮음: {r_sq2}"
|
|
print(f" OK 노이즈 선형: slope={slope2:.3f}, R²={r_sq2:.4f}")
|
|
|
|
# 단일 샘플 (최소 입력)
|
|
slope3, intercept3, r_sq3 = pred_mod.linear_regression([0.0], [5.0])
|
|
assert slope3 == 0.0, "단일샘플 slope 오류"
|
|
print(f" OK 단일 샘플 처리: slope={slope3}")
|
|
|
|
except AssertionError as e:
|
|
print(f" ERR 선형 회귀 수학 오류: {e}")
|
|
ok = False
|
|
except Exception as e:
|
|
print(f" ERR linear_regression 오류: {type(e).__name__}: {e}")
|
|
ok = False
|
|
|
|
print("\n=== 6. predict_value / time_to_reach 검증 ===")
|
|
try:
|
|
# y = 2x + 10, 현재 x=5 (y=20), target y=40 → x=15, delta=10
|
|
slope, intercept = 2.0, 10.0
|
|
pred_val = pred_mod.predict_value(slope, intercept, 5.0)
|
|
assert abs(pred_val - 20.0) < 0.001, f"predict_value 오류: {pred_val}"
|
|
print(f" OK predict_value(x=5) = {pred_val}")
|
|
|
|
ttr = pred_mod.time_to_reach(slope, intercept, 5.0, 40.0)
|
|
assert abs(ttr - 10.0) < 0.001, f"time_to_reach 오류: {ttr}"
|
|
print(f" OK time_to_reach(y=40) = {ttr}시간 후")
|
|
|
|
# 감소 추세에서는 None 반환
|
|
ttr2 = pred_mod.time_to_reach(-1.0, 100.0, 10.0, 150.0)
|
|
assert ttr2 is None, f"감소 추세에서 None이어야 함: {ttr2}"
|
|
print(f" OK 감소 추세 TTR = None (도달 불가)")
|
|
|
|
# slope=0이면 None
|
|
ttr3 = pred_mod.time_to_reach(0.0, 50.0, 0.0, 90.0)
|
|
assert ttr3 is None, f"slope=0에서 None이어야 함: {ttr3}"
|
|
print(f" OK slope=0 TTR = None")
|
|
|
|
except AssertionError as e:
|
|
print(f" ERR TTR 계산 오류: {e}")
|
|
ok = False
|
|
except Exception as e:
|
|
print(f" ERR TTR 오류: {type(e).__name__}: {e}")
|
|
ok = False
|
|
|
|
print("\n=== 7. moving_average 검증 ===")
|
|
try:
|
|
vals = [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0]
|
|
ma = pred_mod.moving_average(vals, window=3)
|
|
assert len(ma) == len(vals), "이동 평균 길이 오류"
|
|
assert abs(ma[2] - 2.0) < 0.001, f"ma[2] 오류: {ma[2]}" # avg(1,2,3)=2
|
|
assert abs(ma[4] - 4.0) < 0.001, f"ma[4] 오류: {ma[4]}" # avg(3,4,5)=4
|
|
print(f" OK 이동 평균(window=3): {[round(v,2) for v in ma]}")
|
|
except AssertionError as e:
|
|
print(f" ERR 이동 평균 오류: {e}")
|
|
ok = False
|
|
except Exception as e:
|
|
print(f" ERR moving_average 오류: {type(e).__name__}: {e}")
|
|
ok = False
|
|
|
|
print("\n=== 8. assess_equipment_lifecycle 검증 ===")
|
|
try:
|
|
from datetime import datetime, timedelta
|
|
|
|
# 수명 초과 장비 (8년 된 서버, 수명 7년)
|
|
old_date = datetime.utcnow() - timedelta(days=365 * 8)
|
|
result = pred_mod.assess_equipment_lifecycle("SERVER", old_date)
|
|
assert result["status"] == "EOL", f"EOL 판정 오류: {result['status']}"
|
|
assert result["usage_pct"] >= 100.0, f"usage_pct 오류: {result['usage_pct']}"
|
|
print(f" OK 8년 서버: status={result['status']}, usage={result['usage_pct']}%")
|
|
|
|
# 경고 단계 (5년 된 서버 → 71% 사용)
|
|
warn_date = datetime.utcnow() - timedelta(days=365 * 5)
|
|
result2 = pred_mod.assess_equipment_lifecycle("SERVER", warn_date)
|
|
assert result2["status"] in ("WARNING", "CRITICAL"), f"경고 판정 오류: {result2['status']}"
|
|
print(f" OK 5년 서버: status={result2['status']}, usage={result2['usage_pct']}%")
|
|
|
|
# 신규 장비 (1년 된 서버)
|
|
new_date = datetime.utcnow() - timedelta(days=365)
|
|
result3 = pred_mod.assess_equipment_lifecycle("SERVER", new_date)
|
|
assert result3["status"] == "HEALTHY", f"HEALTHY 판정 오류: {result3['status']}"
|
|
print(f" OK 1년 서버: status={result3['status']}, usage={result3['usage_pct']}%")
|
|
|
|
# 네트워크 장비 (4년, 수명 5년 → 80%)
|
|
net_date = datetime.utcnow() - timedelta(days=365 * 4)
|
|
result4 = pred_mod.assess_equipment_lifecycle("NETWORK", net_date)
|
|
print(f" OK 4년 네트워크: status={result4['status']}, usage={result4['usage_pct']}%")
|
|
|
|
except AssertionError as e:
|
|
print(f" ERR 수명 주기 평가 오류: {e}")
|
|
ok = False
|
|
except Exception as e:
|
|
print(f" ERR assess_equipment_lifecycle 오류: {type(e).__name__}: {e}")
|
|
ok = False
|
|
|
|
print("\n=== 9. detect_seasonal_pattern 검증 ===")
|
|
try:
|
|
import math
|
|
# 주기 24의 사인파 (뚜렷한 패턴)
|
|
periodic_data = [50 + 20 * math.sin(2 * math.pi * i / 24) for i in range(96)]
|
|
result = pred_mod.detect_seasonal_pattern(periodic_data, period=24)
|
|
assert result["has_pattern"] == True, f"주기성 미감지: {result}"
|
|
print(f" OK 주기성 감지: peak_index={result['peak_index']}, amplitude={result['amplitude']}")
|
|
|
|
# 평탄한 데이터 (패턴 없음)
|
|
flat_data = [50.0 + i * 0.01 for i in range(96)]
|
|
result2 = pred_mod.detect_seasonal_pattern(flat_data, period=24)
|
|
print(f" OK 평탄 데이터: has_pattern={result2['has_pattern']}")
|
|
|
|
# 데이터 부족
|
|
result3 = pred_mod.detect_seasonal_pattern([1.0, 2.0], period=24)
|
|
assert result3["has_pattern"] == False, "데이터 부족 패턴 감지 오류"
|
|
print(f" OK 데이터 부족: has_pattern={result3['has_pattern']}")
|
|
|
|
except AssertionError as e:
|
|
print(f" ERR 계절성 패턴 오류: {e}")
|
|
ok = False
|
|
except Exception as e:
|
|
print(f" ERR detect_seasonal_pattern 오류: {type(e).__name__}: {e}")
|
|
ok = False
|
|
|
|
print("\n=== 10. PREDICTION_THRESHOLDS 구조 검증 ===")
|
|
try:
|
|
thresholds = pred_mod.PREDICTION_THRESHOLDS
|
|
assert isinstance(thresholds, dict), "dict가 아님"
|
|
required_metrics = ["CPU_USAGE", "MEMORY_USAGE", "DISK_USAGE", "RESPONSE_TIME"]
|
|
for mt in required_metrics:
|
|
assert mt in thresholds, f"{mt} 없음"
|
|
cfg = thresholds[mt]
|
|
assert "warning" in cfg, f"{mt}.warning 없음"
|
|
assert "critical" in cfg, f"{mt}.critical 없음"
|
|
assert "unit" in cfg, f"{mt}.unit 없음"
|
|
print(f" OK {mt}: warning={cfg['warning']}, critical={cfg['critical']} {cfg['unit']}")
|
|
|
|
lifespan = pred_mod.EQUIPMENT_LIFESPAN
|
|
assert "SERVER" in lifespan, "SERVER 수명 기준 없음"
|
|
assert "NETWORK" in lifespan, "NETWORK 수명 기준 없음"
|
|
print(f" OK 장비 수명 기준: {lifespan}")
|
|
|
|
except AssertionError as e:
|
|
print(f" ERR 임계값 구조 오류: {e}")
|
|
ok = False
|
|
except Exception as e:
|
|
print(f" ERR 임계값 오류: {type(e).__name__}: {e}")
|
|
ok = False
|
|
|
|
print("\n=== B-6 예측 유지보수 테스트 완료 ===")
|
|
if ok:
|
|
print("모든 검사 통과")
|
|
else:
|
|
sys.exit(1)
|