zioinfo-mail/itsm/test_b6_predictive.py

"""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)