""" test_bot_status_resolver.py — 봇 상태 라벨링 시스템 테스트 task-2375 (dev3/루) 5개 라벨 시나리오 + task-2373 실제 케이스 테스트 """ import json import subprocess import sys import unittest from unittest.mock import patch # scripts 경로를 sys.path에 추가 import os sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "scripts")) import bot_status_resolver as bsr # pyright: ignore[reportMissingImports] # ───────────────────────────────────────────── # 순수 판정 함수 테스트 (classify_status) # ───────────────────────────────────────────── class TestClassifyStatus(unittest.TestCase): """classify_status 순수 함수 5개 라벨 시나리오 테스트.""" def test_alive(self): """ps_alive=True, last_commit_age_min=2 → ALIVE""" result = bsr.classify_status( ps_alive=True, last_commit_age_min=2, pr_merged_at=None, pr_number=None, ) self.assertEqual(result, "ALIVE") def test_idle(self): """ps_alive=True, last_commit_age_min=17 → IDLE""" result = bsr.classify_status( ps_alive=True, last_commit_age_min=17, pr_merged_at=None, pr_number=None, ) self.assertEqual(result, "IDLE") def test_merged(self): """pr_merged_at 있음 → MERGED (다른 조건 무시)""" result = bsr.classify_status( ps_alive=False, last_commit_age_min=100, pr_merged_at="2026-05-02T19:00:00Z", pr_number=42, ) self.assertEqual(result, "MERGED") def test_stale(self): """ps_alive=False, last_commit_age_min=45, pr_number=None → STALE""" result = bsr.classify_status( ps_alive=False, last_commit_age_min=45, pr_merged_at=None, pr_number=None, ) self.assertEqual(result, "STALE") def test_unknown(self): """ps_alive=False, last_commit_age_min=10, pr_number=None → UNKNOWN (dead + 10분 경과 + pr없음 → STALE 조건 age>30 불충족 → UNKNOWN) """ result = bsr.classify_status( ps_alive=False, last_commit_age_min=10, pr_merged_at=None, pr_number=None, ) self.assertEqual(result, "UNKNOWN") def test_merged_takes_priority_over_alive(self): """MERGED는 ALIVE/IDLE보다 우선: ps_alive=True, age=2, merged → MERGED""" result = bsr.classify_status( ps_alive=True, last_commit_age_min=2, pr_merged_at="2026-05-02T19:00:00Z", pr_number=99, ) self.assertEqual(result, "MERGED") def test_alive_boundary_exact_5(self): """경계값: last_commit_age_min=5 (< 5 아님) → IDLE""" result = bsr.classify_status( ps_alive=True, last_commit_age_min=5, pr_merged_at=None, pr_number=None, ) self.assertEqual(result, "IDLE") def test_alive_boundary_4(self): """경계값: last_commit_age_min=4 (< 5) → ALIVE""" result = bsr.classify_status( ps_alive=True, last_commit_age_min=4, pr_merged_at=None, pr_number=None, ) self.assertEqual(result, "ALIVE") # ───────────────────────────────────────────── # verdict 매핑 테스트 # ───────────────────────────────────────────── class TestVerdictMap(unittest.TestCase): def test_all_verdicts(self): expected = { "MERGED": "completed", "ALIVE": "in_progress", "IDLE": "in_progress", "STALE": "stalled", "UNKNOWN": "unknown", } for status, verdict in expected.items(): self.assertEqual(bsr.VERDICT_MAP[status], verdict, f"VERDICT_MAP[{status}] 불일치") # ───────────────────────────────────────────── # resolve() 통합 테스트 (helper mock 기반) # ───────────────────────────────────────────── class TestResolveMocked(unittest.TestCase): """resolve()의 helper 함수들을 mock해서 통합 흐름 테스트.""" def _mock_resolve(self, ps_alive, ps_pids, age_min, sha, pr_number, pr_state, pr_merged_at, markers=None): """공통 mock 패치 유틸.""" if markers is None: markers = {"done": False, "work_done": False, "merged": False, "merge_failed": False} with patch.object(bsr, "_check_ps", return_value=(ps_alive, ps_pids)), \ patch.object(bsr, "_get_last_commit", return_value=(age_min, sha)), \ patch.object(bsr, "_get_pr_info", return_value=(pr_number, pr_state, pr_merged_at)), \ patch.object(bsr, "_check_markers", return_value=markers), \ patch.object(bsr, "_infer_team_from_timers", return_value="dev3"): return bsr.resolve("task-test", "dev3") def test_resolve_alive_status(self): result = self._mock_resolve( ps_alive=True, ps_pids=[12345], age_min=2, sha="abc1234", pr_number=None, pr_state=None, pr_merged_at=None, ) self.assertEqual(result["bot_status"], "ALIVE") self.assertEqual(result["verdict"], "in_progress") self.assertTrue(result["ps_alive"]) def test_resolve_idle_status(self): result = self._mock_resolve( ps_alive=True, ps_pids=[12345], age_min=17, sha="f8e9778", pr_number=None, pr_state=None, pr_merged_at=None, ) self.assertEqual(result["bot_status"], "IDLE") self.assertEqual(result["verdict"], "in_progress") def test_resolve_merged_status(self): result = self._mock_resolve( ps_alive=False, ps_pids=[], age_min=100, sha="aabbcc1", pr_number=42, pr_state="MERGED", pr_merged_at="2026-05-02T19:00:00Z", ) self.assertEqual(result["bot_status"], "MERGED") self.assertEqual(result["verdict"], "completed") self.assertEqual(result["pr_merged_at"], "2026-05-02T19:00:00Z") def test_resolve_stale_status(self): result = self._mock_resolve( ps_alive=False, ps_pids=[], age_min=45, sha="deadbeef", pr_number=None, pr_state=None, pr_merged_at=None, ) self.assertEqual(result["bot_status"], "STALE") self.assertEqual(result["verdict"], "stalled") def test_resolve_unknown_status(self): result = self._mock_resolve( ps_alive=False, ps_pids=[], age_min=10, sha="cafe123", pr_number=None, pr_state=None, pr_merged_at=None, ) self.assertEqual(result["bot_status"], "UNKNOWN") self.assertEqual(result["verdict"], "unknown") # ───────────────────────────────────────────── # task-2373 시뮬레이션 (사고 재현 케이스) # ───────────────────────────────────────────── class TestTask2373Simulation(unittest.TestCase): """2026-05-02 사고 케이스: 페룬(dev6), ps_alive=True (2 PIDs), last_commit_age_min=17, pr_number=None → IDLE + evidence에 no_pr_yet 포함.""" def test_task_2373_simulation(self): markers = {"done": False, "work_done": False, "merged": False, "merge_failed": False} with patch.object(bsr, "_check_ps", return_value=(True, [1356421, 1439822])), \ patch.object(bsr, "_get_last_commit", return_value=(17, "f8e9778")), \ patch.object(bsr, "_get_pr_info", return_value=(None, None, None)), \ patch.object(bsr, "_check_markers", return_value=markers), \ patch.object(bsr, "_infer_team_from_timers", return_value="dev6"): result = bsr.resolve("task-2373", "dev6") # 핵심 검증 self.assertEqual(result["task_id"], "task-2373") self.assertEqual(result["team_id"], "dev6") self.assertEqual(result["bot_status"], "IDLE", "ps가 살아있고 commit이 17분 전이면 IDLE이어야 함") self.assertEqual(result["verdict"], "in_progress") # ps_pids 2개 확인 self.assertEqual(len(result["ps_pids"]), 2) self.assertIn(1356421, result["ps_pids"]) self.assertIn(1439822, result["ps_pids"]) # evidence에 "no_pr_yet" 포함 (PR 아직 없음 = 생성 중) evidence_str = " ".join(result["evidence"]) self.assertTrue( "no_pr_yet" in evidence_str or "PR 생성 중" in evidence_str, f"evidence에 no_pr_yet 없음: {result['evidence']}" ) # commit 정보 확인 self.assertEqual(result["last_commit_age_min"], 17) self.assertEqual(result["last_commit_sha"], "f8e9778") # ───────────────────────────────────────────── # JSON 출력 유효성 테스트 (CLI subprocess) # ───────────────────────────────────────────── class TestJsonOutputValid(unittest.TestCase): """CLI를 subprocess로 호출해서 stdout이 valid JSON인지 확인.""" def test_json_output_valid(self): script = os.path.join( os.path.dirname(__file__), "..", "..", "scripts", "bot_status_resolver.py" ) script = os.path.abspath(script) result = subprocess.run( [sys.executable, script, "--task-id", "task-2375", "--format", "json"], capture_output=True, text=True, timeout=30, ) self.assertEqual(result.returncode, 0, f"CLI 종료 코드 비정상: stderr={result.stderr}") try: data = json.loads(result.stdout) except json.JSONDecodeError as e: self.fail(f"JSON 파싱 실패: {e}\nstdout={result.stdout}") # 필수 키 존재 확인 required_keys = [ "task_id", "team_id", "bot_status", "ps_alive", "ps_pids", "last_commit_age_min", "last_commit_sha", "pr_number", "pr_state", "pr_merged_at", "markers", "verdict", "evidence", ] for key in required_keys: self.assertIn(key, data, f"JSON에 '{key}' 키 없음") # bot_status가 유효한 값 self.assertIn(data["bot_status"], ["MERGED", "ALIVE", "IDLE", "STALE", "UNKNOWN"]) # task_id 일치 self.assertEqual(data["task_id"], "task-2375") # ───────────────────────────────────────────── if __name__ == "__main__": unittest.main(verbosity=2)