#!/usr/bin/env python3 """bot-status-watchdog.py 테스트 스위트""" import json import os import sys import tempfile from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import patch import pytest # 임시 WORKSPACE_ROOT 설정 _TEMP_WORKSPACE = tempfile.mkdtemp() os.environ["WORKSPACE_ROOT"] = _TEMP_WORKSPACE sys.path.insert(0, str(Path(__file__).parent.parent)) watchdog = __import__("bot-status-watchdog") class TestParseSinceTime: """since 시간 파싱 테스트""" def test_parse_utc_z_format(self): """UTC Z 포맷 파싱""" result = watchdog.parse_since_time("2026-03-17T06:00:00Z") assert result is not None assert result.tzinfo == timezone.utc assert result.hour == 6 def test_parse_iso_with_offset(self): """ISO 오프셋 포맷 파싱""" result = watchdog.parse_since_time("2026-03-17T06:00:00+09:00") assert result is not None def test_parse_invalid_returns_none(self): """잘못된 형식은 None 반환""" result = watchdog.parse_since_time("not-a-date") assert result is None def test_parse_empty_returns_none(self): """빈 문자열은 None 반환""" result = watchdog.parse_since_time("") assert result is None class TestLoadBotActivity: """bot-activity.json 로드 테스트""" def test_load_valid_json(self, tmp_path): """유효한 JSON 로드""" bot_file = tmp_path / "bot-activity.json" bot_file.write_text(json.dumps({"bots": {"dev1": {"status": "idle", "since": "2026-03-17T00:00:00Z"}}})) watchdog.BOT_ACTIVITY_FILE = bot_file result = watchdog.load_bot_activity() assert result["bots"]["dev1"]["status"] == "idle" def test_load_missing_file_returns_empty(self, tmp_path): """파일 없으면 빈 dict 반환""" watchdog.BOT_ACTIVITY_FILE = tmp_path / "nonexistent.json" result = watchdog.load_bot_activity() assert result == {"bots": {}} def test_load_invalid_json_returns_empty(self, tmp_path): """잘못된 JSON은 빈 dict 반환""" bot_file = tmp_path / "bot-activity.json" bot_file.write_text("{invalid}") watchdog.BOT_ACTIVITY_FILE = bot_file result = watchdog.load_bot_activity() assert result == {"bots": {}} class TestSaveBotActivity: """bot-activity.json 저장 테스트""" def test_save_creates_file(self, tmp_path): """파일 생성 확인""" bot_file = tmp_path / "bot-activity.json" watchdog.BOT_ACTIVITY_FILE = bot_file data = {"bots": {"dev1": {"status": "idle"}}} result = watchdog.save_bot_activity(data) assert result is True assert bot_file.exists() saved = json.loads(bot_file.read_text()) assert saved["bots"]["dev1"]["status"] == "idle" def test_save_atomic_write(self, tmp_path): """원자적 쓰기 - tmp 파일이 최종 파일로 대체""" bot_file = tmp_path / "bot-activity.json" watchdog.BOT_ACTIVITY_FILE = bot_file data = {"bots": {}} watchdog.save_bot_activity(data) # .tmp 파일이 남아있지 않아야 함 assert not (tmp_path / "bot-activity.tmp").exists() class TestShouldTransitionToIdle: """idle 전환 조건 테스트""" def test_not_timeout_returns_false(self, tmp_path): """30분 미만 → 전환 안 함""" since_time = datetime.now(timezone.utc) - timedelta(minutes=10) with patch.object(watchdog, "find_bot_process", return_value=[]): should, reason = watchdog.should_transition_to_idle("dev1", since_time, 10.0) assert should is False assert reason == "not_timeout" def test_process_alive_returns_false(self, tmp_path): """30분 초과지만 프로세스 살아있음 → 전환 안 함""" since_time = datetime.now(timezone.utc) - timedelta(minutes=40) with patch.object(watchdog, "find_bot_process", return_value=[12345]): should, reason = watchdog.should_transition_to_idle("dev1", since_time, 40.0) assert should is False assert "still_running" in reason assert "12345" in reason def test_done_file_exists_returns_true(self, tmp_path): """프로세스 없음 + .done 파일 있음 → 전환""" since_time = datetime.now(timezone.utc) - timedelta(minutes=40) fake_done = tmp_path / "task-123.done" fake_done.touch() with ( patch.object(watchdog, "find_bot_process", return_value=[]), patch.object(watchdog, "find_recent_done_file", return_value=fake_done), ): should, reason = watchdog.should_transition_to_idle("dev1", since_time, 40.0) assert should is True assert "completed" in reason assert ".done" in reason def test_report_file_exists_returns_true(self, tmp_path): """프로세스 없음 + 보고서 있음 → 전환""" since_time = datetime.now(timezone.utc) - timedelta(minutes=40) fake_report = tmp_path / "task-123.md" fake_report.touch() with ( patch.object(watchdog, "find_bot_process", return_value=[]), patch.object(watchdog, "find_recent_done_file", return_value=None), patch.object(watchdog, "find_recent_report", return_value=fake_report), ): should, reason = watchdog.should_transition_to_idle("dev1", since_time, 40.0) assert should is True assert "completed" in reason assert "보고서" in reason def test_timeout_no_process_no_done_returns_true(self): """프로세스 없음 + .done 없음 + 보고서 없음 + 30분 초과 → 전환""" since_time = datetime.now(timezone.utc) - timedelta(minutes=40) with ( patch.object(watchdog, "find_bot_process", return_value=[]), patch.object(watchdog, "find_recent_done_file", return_value=None), patch.object(watchdog, "find_recent_report", return_value=None), ): should, reason = watchdog.should_transition_to_idle("dev1", since_time, 40.0) assert should is True assert "timeout" in reason class TestFindBotProcess: """find_bot_process 테스트""" def test_unknown_bot_returns_empty(self): """알 수 없는 봇 → 빈 리스트""" result = watchdog.find_bot_process("unknown_bot") assert result == [] def test_pgrep_failure_returns_empty(self): """pgrep 실패 시 빈 리스트 반환""" with patch("subprocess.run", side_effect=OSError("pgrep not found")): result = watchdog.find_bot_process("dev1") assert result == [] def test_pgrep_timeout_returns_empty(self): """pgrep 타임아웃 → 빈 리스트 반환""" import subprocess as _subprocess with patch("subprocess.run", side_effect=_subprocess.TimeoutExpired(["pgrep"], 5)): result = watchdog.find_bot_process("dev1") assert result == [] def test_pgrep_no_match_returns_empty(self): """pgrep 매칭 없음 → 빈 리스트""" mock_result = type("R", (), {"returncode": 1, "stdout": ""})() with patch("subprocess.run", return_value=mock_result): result = watchdog.find_bot_process("dev1") assert result == [] def test_pgrep_returns_pids(self): """pgrep 성공 → PID 목록""" mock_result = type("R", (), {"returncode": 0, "stdout": "12345\n67890\n"})() with patch("subprocess.run", return_value=mock_result): result = watchdog.find_bot_process("dev1") assert result == [12345, 67890] class TestCheckAndRecoverStuckBots: """stuck 봇 자동 복구 테스트""" def test_stuck_bot_recovered(self, tmp_path): """30분 초과 processing 봇 → idle 전환 (프로세스 없음 모킹)""" bot_file = tmp_path / "bot-activity.json" forty_min_ago = (datetime.now(timezone.utc) - timedelta(minutes=40)).strftime("%Y-%m-%dT%H:%M:%SZ") data = {"bots": {"dev1": {"status": "processing", "since": forty_min_ago}}} bot_file.write_text(json.dumps(data)) watchdog.BOT_ACTIVITY_FILE = bot_file watchdog.WATCHDOG_LOG = tmp_path / "watchdog.log" with patch.object(watchdog, "find_bot_process", return_value=[]): recovered = watchdog.check_and_recover_stuck_bots() assert recovered == 1 saved = json.loads(bot_file.read_text()) assert saved["bots"]["dev1"]["status"] == "idle" def test_recent_processing_not_recovered(self, tmp_path): """10분 미만 processing 봇 → 유지""" bot_file = tmp_path / "bot-activity.json" ten_min_ago = (datetime.now(timezone.utc) - timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ") data = {"bots": {"dev2": {"status": "processing", "since": ten_min_ago}}} bot_file.write_text(json.dumps(data)) watchdog.BOT_ACTIVITY_FILE = bot_file watchdog.WATCHDOG_LOG = tmp_path / "watchdog.log" with patch.object(watchdog, "find_bot_process", return_value=[]): recovered = watchdog.check_and_recover_stuck_bots() assert recovered == 0 saved = json.loads(bot_file.read_text()) assert saved["bots"]["dev2"]["status"] == "processing" def test_idle_bot_not_affected(self, tmp_path): """idle 상태 봇은 변경 없음""" bot_file = tmp_path / "bot-activity.json" data = {"bots": {"dev3": {"status": "idle", "since": "2026-01-01T00:00:00Z"}}} bot_file.write_text(json.dumps(data)) watchdog.BOT_ACTIVITY_FILE = bot_file watchdog.WATCHDOG_LOG = tmp_path / "watchdog.log" with patch.object(watchdog, "find_bot_process", return_value=[]): recovered = watchdog.check_and_recover_stuck_bots() assert recovered == 0 def test_watchdog_log_entry_written(self, tmp_path): """복구 시 [WATCHDOG] 로그 기록""" bot_file = tmp_path / "bot-activity.json" log_file = tmp_path / "watchdog.log" forty_min_ago = (datetime.now(timezone.utc) - timedelta(minutes=40)).strftime("%Y-%m-%dT%H:%M:%SZ") data = {"bots": {"dev1": {"status": "processing", "since": forty_min_ago}}} bot_file.write_text(json.dumps(data)) watchdog.BOT_ACTIVITY_FILE = bot_file watchdog.WATCHDOG_LOG = log_file with patch.object(watchdog, "find_bot_process", return_value=[]): watchdog.check_and_recover_stuck_bots() assert log_file.exists() content = log_file.read_text() assert "[WATCHDOG] dev1: processing → idle" in content def test_multiple_stuck_bots_all_recovered(self, tmp_path): """여러 stuck 봇 모두 복구 (프로세스 없음 모킹)""" bot_file = tmp_path / "bot-activity.json" forty_min_ago = (datetime.now(timezone.utc) - timedelta(minutes=40)).strftime("%Y-%m-%dT%H:%M:%SZ") data = { "bots": { "dev1": {"status": "processing", "since": forty_min_ago}, "dev2": {"status": "processing", "since": forty_min_ago}, "dev3": {"status": "idle", "since": "2026-01-01T00:00:00Z"}, } } bot_file.write_text(json.dumps(data)) watchdog.BOT_ACTIVITY_FILE = bot_file watchdog.WATCHDOG_LOG = tmp_path / "watchdog.log" with patch.object(watchdog, "find_bot_process", return_value=[]): recovered = watchdog.check_and_recover_stuck_bots() assert recovered == 2 def test_process_alive_prevents_idle_transition(self, tmp_path): """프로세스 살아있으면 idle 전환 안 됨""" bot_file = tmp_path / "bot-activity.json" forty_min_ago = (datetime.now(timezone.utc) - timedelta(minutes=40)).strftime("%Y-%m-%dT%H:%M:%SZ") data = {"bots": {"dev2": {"status": "processing", "since": forty_min_ago}}} bot_file.write_text(json.dumps(data)) watchdog.BOT_ACTIVITY_FILE = bot_file watchdog.WATCHDOG_LOG = tmp_path / "watchdog.log" with patch.object(watchdog, "find_bot_process", return_value=[99999]): recovered = watchdog.check_and_recover_stuck_bots() assert recovered == 0 saved = json.loads(bot_file.read_text()) assert saved["bots"]["dev2"]["status"] == "processing" def test_invalid_since_skipped(self, tmp_path): """since 파싱 실패한 봇은 스킵""" bot_file = tmp_path / "bot-activity.json" data = {"bots": {"dev1": {"status": "processing", "since": "invalid-date"}}} bot_file.write_text(json.dumps(data)) watchdog.BOT_ACTIVITY_FILE = bot_file watchdog.WATCHDOG_LOG = tmp_path / "watchdog.log" with patch.object(watchdog, "find_bot_process", return_value=[]): recovered = watchdog.check_and_recover_stuck_bots() assert recovered == 0 class TestRunOnce: """1회 실행 테스트""" def test_run_once_no_stuck_bots(self, tmp_path): """stuck 봇 없을 때 정상 실행""" bot_file = tmp_path / "bot-activity.json" bot_file.write_text(json.dumps({"bots": {"dev1": {"status": "idle", "since": "2026-03-17T00:00:00Z"}}})) watchdog.BOT_ACTIVITY_FILE = bot_file watchdog.WATCHDOG_LOG = tmp_path / "watchdog.log" # 예외 없이 실행 확인 watchdog.run_once() assert (tmp_path / "watchdog.log").exists() if __name__ == "__main__": pytest.main([__file__, "-v"])