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