219 lines
9.2 KiB
Python
219 lines
9.2 KiB
Python
"""
|
|
단위 테스트: 자율 비용 최적화 (AutonomousCostOps)
|
|
- cost_optimizer_ai 라우터 로직 검증
|
|
- 낭비 기준 상수 검증
|
|
- AI 텍스트 파싱 로직 검증
|
|
- Pydantic 스키마 검증
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pytest
|
|
from datetime import datetime
|
|
|
|
|
|
# ── 임포트 가능 여부 ──────────────────────────────────────────────────────────
|
|
|
|
def test_module_import():
|
|
"""cost_optimizer_ai 모듈이 오류 없이 임포트되는지 확인."""
|
|
from routers import cost_optimizer_ai
|
|
assert cost_optimizer_ai.router is not None
|
|
|
|
|
|
def test_models_import():
|
|
"""신규 ORM 모델 3개가 임포트되는지 확인."""
|
|
from models import CostAIAnalysis, CostRecommendation, CostForecast
|
|
assert CostAIAnalysis.__tablename__ == "tb_cost_ai_analysis"
|
|
assert CostRecommendation.__tablename__ == "tb_cost_recommendation"
|
|
assert CostForecast.__tablename__ == "tb_cost_forecast"
|
|
|
|
|
|
# ── 라우터 등록 확인 ──────────────────────────────────────────────────────────
|
|
|
|
def test_router_prefix():
|
|
"""라우터 prefix가 /api/cost-ai 인지 확인."""
|
|
from routers.cost_optimizer_ai import router
|
|
assert router.prefix == "/api/cost-ai"
|
|
|
|
|
|
def test_router_endpoints_registered():
|
|
"""8개 엔드포인트가 모두 등록되어 있는지 경로 목록으로 확인."""
|
|
from routers.cost_optimizer_ai import router
|
|
paths = {r.path for r in router.routes}
|
|
assert "/api/cost-ai/analysis" in paths
|
|
assert "/api/cost-ai/analyze" in paths
|
|
assert "/api/cost-ai/forecast/{days}" in paths
|
|
assert "/api/cost-ai/recommendations" in paths
|
|
assert "/api/cost-ai/recommendations/{rec_id}/apply" in paths
|
|
assert "/api/cost-ai/recommendations/{rec_id}/reject" in paths
|
|
assert "/api/cost-ai/waste" in paths
|
|
assert "/api/cost-ai/savings-report" in paths
|
|
|
|
|
|
# ── 낭비 기준 상수 ────────────────────────────────────────────────────────────
|
|
|
|
def test_waste_thresholds():
|
|
"""낭비 감지 기준값이 스펙에 맞는지 확인."""
|
|
from routers.cost_optimizer_ai import (
|
|
_WASTE_CPU_THRESHOLD,
|
|
_WASTE_MEM_THRESHOLD,
|
|
_WASTE_SR_DAYS,
|
|
)
|
|
assert _WASTE_CPU_THRESHOLD == 10.0, "CPU 임계값은 10%"
|
|
assert _WASTE_MEM_THRESHOLD == 20.0, "메모리 임계값은 20%"
|
|
assert _WASTE_SR_DAYS == 30, "SR 미발생 기준은 30일"
|
|
|
|
|
|
# ── 절감 단가 상수 ────────────────────────────────────────────────────────────
|
|
|
|
def test_saving_unit_keys():
|
|
"""절감 단가 딕셔너리에 server, license, cloud 키가 있는지 확인."""
|
|
from routers.cost_optimizer_ai import _SAVING_UNIT
|
|
assert "server" in _SAVING_UNIT
|
|
assert "license" in _SAVING_UNIT
|
|
assert "cloud" in _SAVING_UNIT
|
|
# 모두 양수
|
|
for key, val in _SAVING_UNIT.items():
|
|
assert val > 0, f"{key} 절감 단가는 양수여야 함"
|
|
|
|
|
|
# ── AI 텍스트 파싱 로직 ───────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_recommendations_from_ai_parses_numbered_list():
|
|
"""번호 목록 형식 AI 텍스트가 CostRecommendation 리스트로 변환되는지 확인."""
|
|
from routers.cost_optimizer_ai import _build_recommendations_from_ai
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
mock_db = MagicMock()
|
|
|
|
ai_text = (
|
|
"1. 유휴 서버 통합 가상화\n"
|
|
" CPU 사용률이 낮은 서버를 통합하세요.\n"
|
|
"2. 미사용 라이선스 해지\n"
|
|
" 분기마다 감사를 통해 절감하세요.\n"
|
|
"3. 클라우드 스토리지 티어링\n"
|
|
" cold 티어로 전환하세요.\n"
|
|
)
|
|
recs = await _build_recommendations_from_ai(ai_text, mock_db)
|
|
assert len(recs) == 3, f"3개 권고가 생성되어야 함, 실제: {len(recs)}"
|
|
assert "유휴 서버 통합 가상화" in recs[0].title
|
|
assert "미사용 라이선스 해지" in recs[1].title
|
|
assert "클라우드 스토리지 티어링" in recs[2].title
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_recommendations_limits_to_5():
|
|
"""6개 이상 항목이 있을 때 최대 5개로 제한되는지 확인."""
|
|
from routers.cost_optimizer_ai import _build_recommendations_from_ai
|
|
from unittest.mock import MagicMock
|
|
|
|
mock_db = MagicMock()
|
|
|
|
ai_text = "\n".join(
|
|
f"{i}. 권고 항목 {i}\n 설명 {i}." for i in range(1, 8)
|
|
)
|
|
recs = await _build_recommendations_from_ai(ai_text, mock_db)
|
|
assert len(recs) <= 5, "최대 5개 제한 위반"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_recommendations_empty_text():
|
|
"""빈 AI 텍스트 시 빈 리스트를 반환하는지 확인."""
|
|
from routers.cost_optimizer_ai import _build_recommendations_from_ai
|
|
from unittest.mock import MagicMock
|
|
|
|
mock_db = MagicMock()
|
|
recs = await _build_recommendations_from_ai("", mock_db)
|
|
assert recs == []
|
|
|
|
|
|
# ── Pydantic 스키마 검증 ──────────────────────────────────────────────────────
|
|
|
|
def test_recommendation_out_schema():
|
|
"""RecommendationOut 스키마가 올바른 필드를 갖는지 확인."""
|
|
from routers.cost_optimizer_ai import RecommendationOut
|
|
fields = RecommendationOut.model_fields
|
|
required = {"id", "category", "title", "estimated_saving", "risk_level",
|
|
"auto_applicable", "status", "created_at"}
|
|
for f in required:
|
|
assert f in fields, f"RecommendationOut 필드 누락: {f}"
|
|
|
|
|
|
def test_analysis_out_schema():
|
|
"""AnalysisOut 스키마가 올바른 필드를 갖는지 확인."""
|
|
from routers.cost_optimizer_ai import AnalysisOut
|
|
fields = AnalysisOut.model_fields
|
|
required = {"id", "period", "total_cost", "created_at"}
|
|
for f in required:
|
|
assert f in fields, f"AnalysisOut 필드 누락: {f}"
|
|
|
|
|
|
def test_forecast_out_schema():
|
|
"""ForecastOut 스키마가 올바른 필드를 갖는지 확인."""
|
|
from routers.cost_optimizer_ai import ForecastOut
|
|
fields = ForecastOut.model_fields
|
|
required = {"id", "forecast_date", "predicted_cost", "confidence", "created_at"}
|
|
for f in required:
|
|
assert f in fields, f"ForecastOut 필드 누락: {f}"
|
|
|
|
|
|
# ── ORM 모델 컬럼 검증 ────────────────────────────────────────────────────────
|
|
|
|
def test_cost_ai_analysis_columns():
|
|
"""CostAIAnalysis 모델에 필수 컬럼이 있는지 확인."""
|
|
from models import CostAIAnalysis
|
|
cols = {c.key for c in CostAIAnalysis.__table__.columns}
|
|
required = {"id", "period", "total_cost", "breakdown", "ai_insights",
|
|
"waste_detected", "created_at"}
|
|
for c in required:
|
|
assert c in cols, f"CostAIAnalysis 컬럼 누락: {c}"
|
|
|
|
|
|
def test_cost_recommendation_columns():
|
|
"""CostRecommendation 모델에 필수 컬럼이 있는지 확인."""
|
|
from models import CostRecommendation
|
|
cols = {c.key for c in CostRecommendation.__table__.columns}
|
|
required = {"id", "category", "title", "description", "estimated_saving",
|
|
"risk_level", "auto_applicable", "status", "created_at"}
|
|
for c in required:
|
|
assert c in cols, f"CostRecommendation 컬럼 누락: {c}"
|
|
|
|
|
|
def test_cost_forecast_columns():
|
|
"""CostForecast 모델에 필수 컬럼이 있는지 확인."""
|
|
from models import CostForecast
|
|
cols = {c.key for c in CostForecast.__table__.columns}
|
|
required = {"id", "forecast_date", "predicted_cost", "confidence",
|
|
"factors", "created_at"}
|
|
for c in required:
|
|
assert c in cols, f"CostForecast 컬럼 누락: {c}"
|
|
|
|
|
|
# ── 비즈니스 로직: 예측 수학적 검증 ─────────────────────────────────────────
|
|
|
|
def test_forecast_compound_growth():
|
|
"""복리 성장률 계산 공식이 올바른지 수동 검증."""
|
|
base_cost = 500.0 # 만원
|
|
trend_rate = 0.02 # 월 2%
|
|
month_ahead = 3
|
|
|
|
predicted = base_cost * ((1 + trend_rate) ** month_ahead)
|
|
expected = 500.0 * (1.02 ** 3)
|
|
|
|
assert abs(predicted - expected) < 0.01, f"예측 계산 불일치: {predicted} != {expected}"
|
|
|
|
|
|
def test_confidence_bounds():
|
|
"""신뢰도가 0.3 ~ 0.95 범위 내인지 확인 (예측 공식 검증)."""
|
|
import math
|
|
|
|
def calc_confidence(m: int, history: int) -> float:
|
|
return max(0.3, min(0.95, 0.95 - 0.1 * m - (0.05 if history < 4 else 0)))
|
|
|
|
# 모든 케이스가 범위 내
|
|
for m in range(1, 4):
|
|
for h in (2, 4, 10):
|
|
conf = calc_confidence(m, h)
|
|
assert 0.3 <= conf <= 0.95, f"신뢰도 범위 초과: m={m} h={h} conf={conf}"
|