#!/usr/bin/env python3 """ error_alert.py 단위 테스트 테스트 항목: - check_error_threshold() 정상/비정상 케이스 - format_alert_message() 포맷 검증 - 임계값 미만일 때 False 반환 확인 """ import json import os import sys import tempfile import unittest from datetime import datetime, timedelta from pathlib import Path from unittest.mock import MagicMock, patch # error_alert 모듈 임포트 경로 설정 sys.path.insert(0, str(Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")))) from utils import error_alert def make_error_record(minutes_ago=5, module="test", error_type="ValueError", message="test error"): """ 테스트용 에러 레코드 생성 """ ts = (datetime.now() - timedelta(minutes=minutes_ago)).isoformat() return {"ts": ts, "module": module, "error_type": error_type, "message": message, "traceback": ""} class TestCheckErrorThreshold(unittest.TestCase): """check_error_threshold() 함수 테스트""" def test_no_file_returns_false(self): """파일 없을 때 False 반환""" with patch.object(error_alert, "ERRORS_FILE", "/tmp/nonexistent_file_12345.jsonl"): result = error_alert.check_error_threshold() self.assertFalse(result) def test_below_threshold_returns_false(self): """에러 2개, max_errors=5이면 False""" with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".jsonl") as f: # 최근 에러 2개 작성 for i in range(2): error = make_error_record(minutes_ago=5 + i) f.write(json.dumps(error) + "\n") temp_file = f.name try: with patch.object(error_alert, "ERRORS_FILE", temp_file): result = error_alert.check_error_threshold(minutes=30, max_errors=5) self.assertFalse(result) finally: Path(temp_file).unlink(missing_ok=True) def test_above_threshold_returns_true(self): """에러 6개, max_errors=5이면 True""" with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".jsonl") as f: # 최근 에러 6개 작성 for i in range(6): error = make_error_record(minutes_ago=5 + i) f.write(json.dumps(error) + "\n") temp_file = f.name try: with patch.object(error_alert, "ERRORS_FILE", temp_file): result = error_alert.check_error_threshold(minutes=30, max_errors=5) self.assertTrue(result) finally: Path(temp_file).unlink(missing_ok=True) def test_old_errors_excluded(self): """60분 이전 에러는 카운트 제외""" with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".jsonl") as f: # 5분 전 에러 3개 for i in range(3): error = make_error_record(minutes_ago=5 + i) f.write(json.dumps(error) + "\n") # 70분 전 에러 4개 (제외되어야 함) for i in range(4): error = make_error_record(minutes_ago=70 + i) f.write(json.dumps(error) + "\n") temp_file = f.name try: with patch.object(error_alert, "ERRORS_FILE", temp_file): # 최근 30분 이내만 확인 (5분 전 에러 3개만 카운트) result = error_alert.check_error_threshold(minutes=30, max_errors=5) self.assertFalse(result) finally: Path(temp_file).unlink(missing_ok=True) def test_exact_threshold_returns_false(self): """에러 수 == max_errors이면 False (초과만 True)""" with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".jsonl") as f: # 정확히 5개 에러 작성 for i in range(5): error = make_error_record(minutes_ago=5 + i) f.write(json.dumps(error) + "\n") temp_file = f.name try: with patch.object(error_alert, "ERRORS_FILE", temp_file): result = error_alert.check_error_threshold(minutes=30, max_errors=5) self.assertFalse(result) finally: Path(temp_file).unlink(missing_ok=True) class TestGetRecentErrorsInWindow(unittest.TestCase): """get_recent_errors_in_window() 함수 테스트""" def test_returns_recent_only(self): """30분 이내 에러만 반환""" with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".jsonl") as f: # 5분 전 에러 2개 recent_errors = [] for i in range(2): error = make_error_record(minutes_ago=5 + i) f.write(json.dumps(error) + "\n") recent_errors.append(error) # 70분 전 에러 1개 (제외되어야 함) old_error = make_error_record(minutes_ago=70) f.write(json.dumps(old_error) + "\n") temp_file = f.name try: with patch.object(error_alert, "ERRORS_FILE", temp_file): result = error_alert.get_recent_errors_in_window(minutes=30) self.assertEqual(len(result), 2) finally: Path(temp_file).unlink(missing_ok=True) def test_empty_file_returns_empty(self): """빈 파일이면 []""" with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".jsonl") as f: temp_file = f.name try: with patch.object(error_alert, "ERRORS_FILE", temp_file): result = error_alert.get_recent_errors_in_window(minutes=30) self.assertEqual(result, []) finally: Path(temp_file).unlink(missing_ok=True) class TestFormatAlertMessage(unittest.TestCase): """format_alert_message() 함수 테스트""" def test_format_contains_alert_header(self): """[에러 알림] 포함""" errors = [make_error_record(minutes_ago=5)] result = error_alert.format_alert_message(errors) self.assertIn("[에러 알림]", result) def test_format_shows_count(self): """에러 수 포함""" errors = [make_error_record(minutes_ago=5) for _ in range(3)] result = error_alert.format_alert_message(errors) self.assertIn("3건", result) def test_format_max_3_items(self): """5개 에러 중 최대 3개 표시""" errors = [make_error_record(minutes_ago=5 + i, message=f"error {i}") for i in range(5)] result = error_alert.format_alert_message(errors) # 3개까지만 표시되어야 함 self.assertIn("error 0", result) self.assertIn("error 1", result) self.assertIn("error 2", result) # 4번째, 5번째는 "외 N건"으로 표시되어야 함 self.assertNotIn("error 3", result) self.assertNotIn("error 4", result) def test_format_shows_more_indicator(self): """3개 초과 시 "외 N건" 포함""" errors = [make_error_record(minutes_ago=5 + i) for i in range(5)] result = error_alert.format_alert_message(errors) self.assertIn("외 2건", result) def test_empty_errors_handled(self): """빈 리스트도 처리 가능""" errors = [] result = error_alert.format_alert_message(errors) self.assertIn("[에러 알림]", result) self.assertIsInstance(result, str) class TestSendAlert(unittest.TestCase): """send_alert() 함수 테스트""" @patch("utils.error_alert.subprocess.run") def test_send_success(self, mock_run): """subprocess.run 성공 시 True""" mock_run.return_value = MagicMock(returncode=0) result = error_alert.send_alert("test message") self.assertTrue(result) mock_run.assert_called_once() @patch("utils.error_alert.subprocess.run") def test_send_failure(self, mock_run): """subprocess.run 실패 시 False (예외 없이)""" mock_run.return_value = MagicMock(returncode=1) result = error_alert.send_alert("test message") self.assertFalse(result) @patch("utils.error_alert.subprocess.run") def test_send_exception_returns_false(self, mock_run): """subprocess.run 예외 발생 시 False""" mock_run.side_effect = Exception("Connection failed") result = error_alert.send_alert("test message") self.assertFalse(result) if __name__ == "__main__": unittest.main()