"""task-2422 Fix B 회귀 테스트: 신호등 봇 실 PID 종합 판정 검증 포인트: - evaluate_task_liveness 6 시나리오 (completed/no-hb/fresh-pid-alive/fresh-pid-dead/stale/escalated-status박제) - _PidLivenessProvider override 동작 - get_running_tasks_by_team이 status=escalated여도 PID alive 시 작업중 표시 (사고 회귀 차단) - task-2414/2417/2420 사고 재현: status가 escalated로 박제되어도 신호등 작업중 정확 """ import os import sys import tempfile import time from pathlib import Path import pytest # project import if "/home/jay/workspace" not in sys.path: sys.path.insert(0, "/home/jay/workspace") from dashboard.data_loader import ( # type: ignore[import-not-found] # noqa: E402 DataLoader, HEARTBEAT_FRESH_SECONDS, _PidLivenessProvider, evaluate_task_liveness, ) @pytest.fixture def tmp_heartbeat_dir(): with tempfile.TemporaryDirectory() as tmp: yield Path(tmp) def _touch_fresh(hb_dir: Path, task_id: str): """방금 mtime의 heartbeat 파일 생성""" f = hb_dir / f"{task_id}.heartbeat" f.touch() return f def _touch_stale(hb_dir: Path, task_id: str, age_sec=HEARTBEAT_FRESH_SECONDS + 60): """오래된 mtime의 heartbeat 파일 생성""" f = hb_dir / f"{task_id}.heartbeat" f.touch() past = time.time() - age_sec os.utime(f, (past, past)) return f def test_liveness_completed_status_returns_idle(tmp_heartbeat_dir): """status=completed → idle (heartbeat/PID 무관)""" _touch_fresh(tmp_heartbeat_dir, "t1") p = _PidLivenessProvider() p.set_override({"sched-1": True}) result = evaluate_task_liveness( "t1", {"status": "completed", "schedule_id": "sched-1"}, tmp_heartbeat_dir, p ) assert result == "idle" def test_liveness_running_no_heartbeat_returns_idle(tmp_heartbeat_dir): """status=running, heartbeat 파일 없음 → idle""" p = _PidLivenessProvider() p.set_override({"sched-2": True}) result = evaluate_task_liveness( "t2", {"status": "running", "schedule_id": "sched-2"}, tmp_heartbeat_dir, p ) assert result == "idle" def test_liveness_running_fresh_heartbeat_pid_alive_returns_working(tmp_heartbeat_dir): """status=running + fresh heartbeat + PID alive → working""" _touch_fresh(tmp_heartbeat_dir, "t3") p = _PidLivenessProvider() p.set_override({"sched-3": True}) result = evaluate_task_liveness( "t3", {"status": "running", "schedule_id": "sched-3"}, tmp_heartbeat_dir, p ) assert result == "working" def test_liveness_running_fresh_heartbeat_pid_dead_returns_bot_suspect_dead(tmp_heartbeat_dir): """status=running + fresh heartbeat + PID dead → bot_suspect_dead""" _touch_fresh(tmp_heartbeat_dir, "t4") p = _PidLivenessProvider() p.set_override({}) # 비어있음 → 모든 sched가 dead result = evaluate_task_liveness( "t4", {"status": "running", "schedule_id": "sched-4"}, tmp_heartbeat_dir, p ) assert result == "bot_suspect_dead" def test_liveness_running_stale_heartbeat_returns_idle(tmp_heartbeat_dir): """status=running + stale heartbeat → idle (PID alive여도)""" _touch_stale(tmp_heartbeat_dir, "t5") p = _PidLivenessProvider() p.set_override({"sched-5": True}) result = evaluate_task_liveness( "t5", {"status": "running", "schedule_id": "sched-5"}, tmp_heartbeat_dir, p ) assert result == "idle" def test_liveness_escalated_status_with_alive_signals_returns_working(tmp_heartbeat_dir): """★ 핵심: status=escalated여도 fresh heartbeat + PID alive → working (status박제 무시) task-2414/2417/2420 사고 회귀 차단의 핵심 시나리오. """ _touch_fresh(tmp_heartbeat_dir, "t6") p = _PidLivenessProvider() p.set_override({"sched-6": True}) result = evaluate_task_liveness( "t6", {"status": "escalated", "schedule_id": "sched-6"}, tmp_heartbeat_dir, p ) assert result == "working", f"status박제에 휘둘림 (회귀): expected working, got {result}" def test_liveness_unknown_status_returns_idle(tmp_heartbeat_dir): """status=foobar 등 알수없는 값 → idle""" _touch_fresh(tmp_heartbeat_dir, "t7") p = _PidLivenessProvider() p.set_override({"sched-7": True}) result = evaluate_task_liveness( "t7", {"status": "foobar", "schedule_id": "sched-7"}, tmp_heartbeat_dir, p ) assert result == "idle" def test_pid_provider_override_returns_false_for_missing_schedule(): """override map에 없는 schedule_id → False""" p = _PidLivenessProvider() p.set_override({"sched-A": True}) assert p.is_alive("sched-B") is False assert p.is_alive("") is False assert p.is_alive(None) is False def test_pid_provider_clear_override_resets_to_real_lookup(): """set_override(None) → 실제 pgrep 모드 복귀 (캐시 reset)""" p = _PidLivenessProvider() p.set_override({"sched-X": True}) assert p.is_alive("sched-X") is True p.set_override(None) # 실제 pgrep는 schedule "sched-X" 못 찾을 것 → False assert p.is_alive("sched-X") is False def test_get_running_tasks_by_team_with_escalated_status_includes_alive_bot(): """★ 통합 테스트: status=escalated + fresh heartbeat + PID alive → 신호등에 작업중 표시""" with tempfile.TemporaryDirectory() as tmp: workspace = Path(tmp) memory = workspace / "memory" (memory / "heartbeats").mkdir(parents=True) loader = DataLoader(workspace) # heartbeat 파일 작성 (fresh) (memory / "heartbeats" / "task-9999.heartbeat").touch() # task_data 직접 주입 loader.task_data = { "tasks": { "task-9999": { "status": "escalated", # 박제된 상태 "team_id": "dev2-team", "schedule_id": "sched-test-9999", "description": "test task", "start_time": "2026-05-01T00:00:00", "bot": "bot-x", } } } # PID alive override loader.pid_provider.set_override({"sched-test-9999": True}) result = loader.get_running_tasks_by_team() assert "dev2-team" in result, f"escalated + alive 봇이 신호등에서 누락 (회귀): {result}" assert len(result["dev2-team"]) == 1 entry = result["dev2-team"][0] assert entry["task_id"] == "task-9999" assert entry.get("liveness") == "working" def test_get_running_tasks_by_team_excludes_running_with_dead_bot(): """status=running이지만 heartbeat stale → 신호등에서 제외""" with tempfile.TemporaryDirectory() as tmp: workspace = Path(tmp) memory = workspace / "memory" (memory / "heartbeats").mkdir(parents=True) loader = DataLoader(workspace) # stale heartbeat hb = memory / "heartbeats" / "task-9998.heartbeat" hb.touch() past = time.time() - HEARTBEAT_FRESH_SECONDS - 100 os.utime(hb, (past, past)) loader.task_data = { "tasks": { "task-9998": { "status": "running", "team_id": "dev1-team", "schedule_id": "sched-9998", "description": "stale task", } } } loader.pid_provider.set_override({"sched-9998": False}) result = loader.get_running_tasks_by_team() assert "dev1-team" not in result, "stale heartbeat task가 신호등에 포함됨" def test_get_running_tasks_by_team_marks_bot_suspect_dead(): """fresh heartbeat + PID dead → entry liveness=bot_suspect_dead""" with tempfile.TemporaryDirectory() as tmp: workspace = Path(tmp) memory = workspace / "memory" (memory / "heartbeats").mkdir(parents=True) loader = DataLoader(workspace) (memory / "heartbeats" / "task-7777.heartbeat").touch() loader.task_data = { "tasks": { "task-7777": { "status": "running", "team_id": "dev3-team", "schedule_id": "sched-7777", "description": "bot crashed", } } } loader.pid_provider.set_override({}) # 모두 dead result = loader.get_running_tasks_by_team() assert "dev3-team" in result assert result["dev3-team"][0]["liveness"] == "bot_suspect_dead" def test_data_loader_initializes_pid_provider(): """DataLoader 인스턴스에 pid_provider 속성 존재""" with tempfile.TemporaryDirectory() as tmp: loader = DataLoader(Path(tmp)) assert hasattr(loader, "pid_provider") assert isinstance(loader.pid_provider, _PidLivenessProvider) assert hasattr(loader, "heartbeat_dir") assert loader.heartbeat_dir == Path(tmp) / "memory" / "heartbeats"