"""A-2 SLA 구문 + 임포트 + 단위 테스트""" import sys, ast, os # ── 1. 구문 검사 ───────────────────────────────────────────────────────────── files = [ "core/sla.py", "routers/tasks.py", "models.py", "core/scheduler.py", ] print("=== 1. 구문 검사 ===") 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) # ── 2. SLA 순수 로직 단위 테스트 (DB 불필요) ────────────────────────────── print("\n=== 2. SLA 단위 테스트 ===") from datetime import datetime, timedelta os.environ.setdefault("GUARDIA_SECRET_KEY", "test-secret-key-32bytes-padding!!") os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_sla.db") from core.sla import ( compute_sla_deadline, is_sla_breached, sla_remaining_minutes, ) # compute_sla_deadline now = datetime(2026, 1, 1, 9, 0, 0) cases = [ ("CRITICAL", 4, 2.0), # 4 * 0.5 = 2h ("HIGH", 4, 3.0), # 4 * 0.75 = 3h ("MEDIUM", 4, 4.0), # 4 * 1.0 = 4h ("LOW", 4, 8.0), # 4 * 2.0 = 8h ] for priority, sla_h, expected_h in cases: dl = compute_sla_deadline(now, sla_h, priority) diff = (dl - now).total_seconds() / 3600 status = "OK" if abs(diff - expected_h) < 0.001 else "FAIL" print(f" {status} compute_sla_deadline({priority}, sla={sla_h}h) → {diff}h (expect {expected_h}h)") # is_sla_breached past = datetime.now() - timedelta(minutes=10) future = datetime.now() + timedelta(minutes=10) assert is_sla_breached(past) == True, "breached(past) should be True" assert is_sla_breached(future) == False, "breached(future) should be False" assert is_sla_breached(None) == False, "breached(None) should be False" print(" OK is_sla_breached (past/future/None)") # sla_remaining_minutes remaining = sla_remaining_minutes(future) assert remaining is not None and remaining > 0, f"remaining should be > 0, got {remaining}" overdue = sla_remaining_minutes(past) assert overdue is not None and overdue < 0, f"overdue should be < 0, got {overdue}" assert sla_remaining_minutes(None) is None, "None deadline → None" print(f" OK sla_remaining_minutes (future={remaining}m, past={overdue}m, None=None)") # ── 3. SROut 스키마 필드 확인 ──────────────────────────────────────────────── print("\n=== 3. SROut 스키마 SLA 필드 확인 ===") from models import SROut import inspect fields = SROut.model_fields if hasattr(SROut, "model_fields") else SROut.__fields__ sla_fields = ["sla_deadline", "sla_breached", "escalated_at", "escalated_to"] for field in sla_fields: if field in fields: print(f" OK SROut.{field} exists") else: print(f" ERR SROut.{field} MISSING") ok = False # ── 4. tasks.py SLA 엔드포인트 라우트 확인 ────────────────────────────────── print("\n=== 4. tasks.py SLA 엔드포인트 라우트 확인 ===") import importlib.util, types spec = importlib.util.spec_from_file_location("tasks_mod", "routers/tasks.py") tasks_mod = importlib.util.module_from_spec(spec) # 라우터 객체만 임포트 (DB 연결 없이) try: spec.loader.exec_module(tasks_mod) router = tasks_mod.router routes = {r.path: [m for m in r.methods] for r in router.routes if hasattr(r, "methods")} target_routes = [ "/api/tasks/{sr_id}/sla", "/api/tasks/sla/violations", ] for path in target_routes: if any(path.lstrip("/api/tasks") in r or r == path.replace("/api/tasks", "") for r in routes): print(f" OK 경로 존재: {path}") else: # prefix 제거 후 실제 경로 확인 short = path.replace("/api/tasks", "") found = any(short in r for r in routes.keys()) status = "OK" if found else "WARN" print(f" {status} 경로: {short} → routes={list(routes.keys())[:5]}") except Exception as e: print(f" INFO 라우터 로드 중 외부 의존성 에러 (정상): {type(e).__name__}: {str(e)[:80]}") print("\n=== 테스트 완료 ===") if ok: print("모든 테스트 통과 ✓") else: print("일부 테스트 실패 ✗") sys.exit(1)