zioinfo-mail/itsm/test_a4_timeline.py

"""A-4 운영 이벤트 타임라인 테스트"""
import sys, ast, os, asyncio, json
from datetime import datetime, timedelta

os.environ.setdefault("GUARDIA_SECRET_KEY", "test-timeline-secret-key-32bytes!")
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_timeline.db")

print("=== 1. 구문 검사 ===")
files = ["routers/timeline.py", "main.py"]
ok = True
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
if not ok:
    sys.exit(1)

print("\n=== 2. 모듈 임포트 검사 ===")
try:
    import importlib.util
    spec = importlib.util.spec_from_file_location("timeline_mod", "routers/timeline.py")
    timeline_mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(timeline_mod)

    # EVENT_TYPES 확인
    ev = timeline_mod.EVENT_TYPES
    expected = {
        "sr_created", "sr_status_changed", "sr_sla_violated", "sr_escalated",
        "deploy_started", "deploy_completed", "deploy_failed",
        "batch_started", "batch_completed", "batch_failed",
        "oncall_assigned", "incident_created", "incident_resolved",
    }
    assert ev == expected, f"EVENT_TYPES mismatch: {ev}"
    print(f"  OK  EVENT_TYPES = {len(ev)}개 정의")

    # 라우터 존재 확인
    assert hasattr(timeline_mod, "router"), "router 없음"
    print(f"  OK  router 객체 존재")

    # 헬퍼 함수 존재 확인
    for fn in ["_collect_sr_events", "_collect_audit_events",
               "_collect_deploy_events", "_collect_batch_events"]:
        assert hasattr(timeline_mod, fn), f"{fn} 없음"
    print(f"  OK  이벤트 수집 헬퍼 4개 존재")

    # 엔드포인트 경로 확인
    routes = {r.path for r in timeline_mod.router.routes}
    assert "/api/timeline" in routes, f"GET /api/timeline 없음: {routes}"
    assert "/api/timeline/summary" in routes, f"GET /api/timeline/summary 없음: {routes}"
    print(f"  OK  엔드포인트: {sorted(routes)}")

except Exception as e:
    print(f"  ERR 임포트 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 3. 이벤트 수집 헬퍼 단위 테스트 ===")

async def test_collect_helpers():
    """DB 없이 필터 타입 로직만 테스트."""
    from routers.timeline import _collect_sr_events, _collect_audit_events, \
        _collect_deploy_events, _collect_batch_events, EVENT_TYPES

    start = datetime.now() - timedelta(days=7)
    end   = datetime.now()

    # 빈 filter_types 시 빈 결과 반환 확인
    class MockDB:
        async def execute(self, *a, **kw):
            raise Exception("DB should not be called")

    # sr_created 미포함 필터 → DB 호출 없이 빈 결과
    result = await _collect_sr_events(MockDB(), start, end, {"deploy_started"})
    assert result == [], f"Expected [] when no sr types in filter, got {result}"
    print(f"  OK  sr 이벤트: 필터 미포함 시 DB 미조회")

    # sr_status_changed 미포함 필터 → 빈 결과
    result = await _collect_audit_events(MockDB(), start, end, {"sr_created"})
    assert result == [], f"Expected [] for audit without sr_status_changed"
    print(f"  OK  audit 이벤트: 필터 미포함 시 DB 미조회")

    # deploy 타입 미포함 필터 → 빈 결과
    result = await _collect_deploy_events(MockDB(), start, end, {"sr_created"})
    assert result == [], f"Expected [] for deploy without deploy types"
    print(f"  OK  deploy 이벤트: 필터 미포함 시 DB 미조회")

    # batch 타입 미포함 필터 → 빈 결과
    result = await _collect_batch_events(MockDB(), start, end, {"sr_created"})
    assert result == [], f"Expected [] for batch without batch types"
    print(f"  OK  batch 이벤트: 필터 미포함 시 DB 미조회")

asyncio.run(test_collect_helpers())

print("\n=== 4. 이벤트 정렬 및 페이지네이션 로직 테스트 ===")

def test_pagination_logic():
    """실제 이벤트 목록을 시뮬레이션하여 정렬/페이지네이션 검증."""
    events = []
    base = datetime(2024, 1, 10, 12, 0, 0)
    for i in range(15):
        ts = base - timedelta(hours=i)
        events.append({
            "id": f"sr_created_{i}",
            "type": "sr_created",
            "timestamp": ts.isoformat(),
            "title": f"SR {i}",
            "priority": "HIGH" if i % 2 == 0 else "LOW",
        })

    # 시간 역순 정렬
    events.sort(key=lambda e: e["timestamp"], reverse=True)
    assert events[0]["id"] == "sr_created_0", "최신 이벤트가 첫 번째여야 함"
    assert events[-1]["id"] == "sr_created_14", "가장 오래된 이벤트가 마지막이어야 함"
    print("  OK  시간 역순 정렬")

    # 우선순위 필터
    high_only = [e for e in events if e.get("priority") == "HIGH"]
    assert len(high_only) == 8, f"HIGH 우선순위 8개 예상, 실제: {len(high_only)}"
    print(f"  OK  우선순위 필터 (HIGH: {len(high_only)}개)")

    # 페이지네이션
    total = len(events)  # 15
    skip, limit = 0, 5
    page1 = events[skip:skip + limit]
    assert len(page1) == 5
    assert (skip + limit) < total  # has_more = True
    print(f"  OK  페이지1 (skip=0, limit=5): {len(page1)}개, has_more=True")

    skip2, limit2 = 10, 5
    page2 = events[skip2:skip2 + limit2]
    assert len(page2) == 5
    assert not ((skip2 + limit2) < total)  # has_more = False (10+5=15 = total)
    print(f"  OK  페이지3 (skip=10, limit=5): {len(page2)}개, has_more=False")

test_pagination_logic()

print("\n=== 5. 이벤트 구조 검증 ===")

def test_event_structure():
    """이벤트 딕셔너리 필수 키 검증."""
    required_keys = {"id", "type", "timestamp", "title", "detail", "priority",
                     "ref_id", "actor", "icon", "color"}

    # SR created 이벤트 시뮬레이션
    sr_event = {
        "id":        "sr_created_SR-001",
        "type":      "sr_created",
        "timestamp": datetime.now().isoformat(),
        "title":     "SR 접수: 시스템 오류",
        "detail":    "우선순위: HIGH | 담당: 미배정",
        "priority":  "HIGH",
        "ref_id":    "SR-001",
        "actor":     "user1",
        "icon":      "ticket",
        "color":     "#2563eb",
    }
    missing = required_keys - sr_event.keys()
    assert not missing, f"SR 이벤트 누락 키: {missing}"
    print("  OK  SR 이벤트 구조 (필수 키 10개)")

    # SLA 위반 이벤트
    sla_event = {
        "id":        "sr_sla_violated_SR-001",
        "type":      "sr_sla_violated",
        "timestamp": datetime.now().isoformat(),
        "title":     "SLA 위반: 시스템 오류",
        "detail":    "담당: 미배정 | 에스컬레이션: 없음",
        "priority":  "HIGH",
        "ref_id":    "SR-001",
        "actor":     "SYSTEM",
        "icon":      "alert",
        "color":     "#dc2626",
    }
    missing = required_keys - sla_event.keys()
    assert not missing, f"SLA 이벤트 누락 키: {missing}"
    print("  OK  SLA 위반 이벤트 구조")

    # 배포 이벤트
    deploy_event = {
        "id":        "deploy_start_1",
        "type":      "deploy_started",
        "timestamp": datetime.now().isoformat(),
        "title":     "배포 시작: 세션 #1",
        "detail":    "SR: SR-001 | 시작: admin",
        "priority":  None,
        "ref_id":    "1",
        "actor":     "admin",
        "icon":      "rocket",
        "color":     "#7c3aed",
    }
    missing = required_keys - deploy_event.keys()
    assert not missing, f"배포 이벤트 누락 키: {missing}"
    print("  OK  배포 이벤트 구조")

test_event_structure()

print("\n=== 6. 요약 집계 로직 테스트 ===")

def test_summary_logic():
    """일별 카운트 집계 로직 검증."""
    days = 7
    end = datetime.now()
    start = end - timedelta(days=days)

    # by_day 초기화
    by_day = {}
    for d_offset in range(days):
        d = (end - timedelta(days=d_offset)).date()
        by_day[d.isoformat()] = {"sr": 0, "deploy": 0, "sla_violation": 0}

    assert len(by_day) == 7, f"7일치 슬롯 필요, 실제: {len(by_day)}"
    print(f"  OK  7일치 집계 슬롯 초기화")

    # 테스트 SR 데이터 집계
    today_key = datetime.now().date().isoformat()
    by_day[today_key]["sr"] += 3
    by_day[today_key]["sla_violation"] += 1

    yesterday_key = (datetime.now().date() - timedelta(days=1)).isoformat()
    by_day[yesterday_key]["sr"] += 2
    by_day[yesterday_key]["deploy"] += 5

    totals = {
        "sr":            sum(v["sr"] for v in by_day.values()),
        "deploy":        sum(v["deploy"] for v in by_day.values()),
        "sla_violation": sum(v["sla_violation"] for v in by_day.values()),
    }
    assert totals["sr"] == 5, f"SR 합계 5 예상, 실제: {totals['sr']}"
    assert totals["deploy"] == 5, f"Deploy 합계 5 예상, 실제: {totals['deploy']}"
    assert totals["sla_violation"] == 1, f"SLA 위반 합계 1 예상"
    print(f"  OK  일별 카운트 집계: SR={totals['sr']}, Deploy={totals['deploy']}, SLA={totals['sla_violation']}")

    # sorted_days (오름차순)
    sorted_days = sorted(by_day.items())
    dates_list = [d for d, _ in sorted_days]
    assert dates_list == sorted(dates_list), "날짜 오름차순 정렬 필요"
    print(f"  OK  날짜 오름차순 정렬")

test_summary_logic()

print("\n=== 7. main.py 등록 확인 ===")
with open("main.py", encoding="utf-8") as f:
    main_src = f.read()

checks = [
    ("timeline", "timeline 라우터 임포트"),
    ("timeline.router", "timeline 라우터 등록"),
]
for sym, desc in checks:
    status = "OK" if sym in main_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc} ({sym})")

print("\n=== A-4 타임라인 테스트 완료 ===")
if ok:
    print("모든 검사 통과")
else:
    sys.exit(1)