""" tests/test_session_resilience.py SessionResilience 단위 테스트 (TDD - RED 단계) 테스트 항목: - test_init_defaults: 초기화 시 기본 경로 확인 - test_check_all_sessions_no_running: 실행 중인 세션 없을 때 빈 결과 - test_check_all_sessions_normal_level: normal 레벨 세션은 actions 없음 - test_check_all_sessions_warning_level: warning 레벨 감지 시 이벤트 기록 - test_check_all_sessions_critical_level: critical 레벨 감지 시 이벤트 + resume 트리거 - test_check_session_returns_status: 개별 세션 체크 결과 형식 확인 - test_handle_warning_creates_event_file: WARNING 이벤트 파일 생성 확인 - test_handle_critical_creates_event_file: CRITICAL 이벤트 파일 생성 확인 - test_handle_critical_saves_session_summary: CRITICAL 시 세션 요약 저장 확인 - test_handle_critical_triggers_resume: CRITICAL 시 resume 트리거 (dispatch 호출 mock) - test_already_handled_session_not_re_triggered: 이미 처리된 세션 중복 트리거 방지 - test_get_bot_mapping: 봇 ID ↔ 팀 매핑 확인 """ from __future__ import annotations import json import os import sys from pathlib import Path from unittest.mock import MagicMock, patch import pytest _WORKSPACE = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(_WORKSPACE) not in sys.path: sys.path.insert(0, str(_WORKSPACE)) # --------------------------------------------------------------------------- # 헬퍼: 테스트용 데이터 생성 # --------------------------------------------------------------------------- def _make_timers(tasks: dict) -> dict: return {"tasks": tasks} def _make_ledger(tasks: dict) -> dict: return {"tasks": tasks} def _make_running_task(task_id: str, team_id: str) -> dict: return { "task_id": task_id, "team_id": team_id, "status": "running", } def _make_ledger_entry(total_tokens: int, team_id: str = "dev2-team") -> dict: return { "team_id": team_id, "total_tokens": total_tokens, "input_tokens": total_tokens // 2, "output_tokens": total_tokens // 2, } # --------------------------------------------------------------------------- # 1. TestSessionResilience 클래스 # --------------------------------------------------------------------------- class TestSessionResilience: """SessionResilience 오케스트레이터 테스트""" def test_init_defaults(self, tmp_path: Path) -> None: """초기화 시 기본 경로 확인""" from utils.session_resilience import SessionResilience sr = SessionResilience(workspace_root=str(tmp_path)) # 기본 파라미터 확인 assert sr.context_limit == 200_000 assert sr.warning_pct == 0.70 assert sr.critical_pct == 0.85 # 경로 확인 assert sr.timers_path == tmp_path / "memory" / "task-timers.json" assert sr.ledger_path == tmp_path / "memory" / "token-ledger.json" assert sr.events_dir == tmp_path / "memory" / "events" assert sr.sessions_dir == tmp_path / "memory" / "sessions" def test_check_all_sessions_no_running(self, tmp_path: Path) -> None: """실행 중인 세션 없을 때 빈 결과 반환""" from utils.session_resilience import SessionResilience # task-timers.json: 완료된 작업만 존재 timers_data = _make_timers( { "task-100.1": { **_make_running_task("task-100.1", "dev2-team"), "status": "completed", } } ) ledger_data = _make_ledger({}) (tmp_path / "memory").mkdir(parents=True) (tmp_path / "memory" / "task-timers.json").write_text(json.dumps(timers_data)) (tmp_path / "memory" / "token-ledger.json").write_text(json.dumps(ledger_data)) sr = SessionResilience(workspace_root=str(tmp_path)) result = sr.check_all_sessions() assert result["checked"] == 0 assert result["warnings"] == [] assert result["criticals"] == [] assert result["normals"] == 0 assert "timestamp" in result def test_check_all_sessions_normal_level(self, tmp_path: Path) -> None: """normal 레벨 세션은 warnings/criticals 리스트에 없음""" from utils.session_resilience import SessionResilience # 50% 사용 = normal total_tokens = 100_000 # 100_000 / 200_000 = 50% task_id = "task-200.1" team_id = "dev2-team" timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(total_tokens, team_id)}) (tmp_path / "memory").mkdir(parents=True) (tmp_path / "memory" / "task-timers.json").write_text(json.dumps(timers_data)) (tmp_path / "memory" / "token-ledger.json").write_text(json.dumps(ledger_data)) sr = SessionResilience(workspace_root=str(tmp_path)) result = sr.check_all_sessions() assert result["checked"] == 1 assert result["warnings"] == [] assert result["criticals"] == [] assert result["normals"] == 1 def test_check_all_sessions_warning_level(self, tmp_path: Path) -> None: """warning 레벨 감지 시 이벤트 기록 및 warnings 리스트에 포함""" from utils.session_resilience import SessionResilience # 75% 사용 = warning total_tokens = 150_000 # 150_000 / 200_000 = 75% task_id = "task-300.1" team_id = "dev2-team" timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(total_tokens, team_id)}) (tmp_path / "memory").mkdir(parents=True) (tmp_path / "memory" / "task-timers.json").write_text(json.dumps(timers_data)) (tmp_path / "memory" / "token-ledger.json").write_text(json.dumps(ledger_data)) sr = SessionResilience(workspace_root=str(tmp_path)) result = sr.check_all_sessions() assert result["checked"] == 1 assert len(result["warnings"]) == 1 assert result["criticals"] == [] assert result["normals"] == 0 warning = result["warnings"][0] assert warning["task_id"] == task_id assert "usage_pct" in warning assert "event_path" in warning def test_check_all_sessions_critical_level(self, tmp_path: Path) -> None: """critical 레벨 감지 시 이벤트 기록 + resume 트리거""" from utils.session_resilience import SessionResilience # 90% 사용 = critical total_tokens = 180_000 # 180_000 / 200_000 = 90% task_id = "task-400.1" team_id = "dev2-team" timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(total_tokens, team_id)}) (tmp_path / "memory").mkdir(parents=True) (tmp_path / "memory" / "task-timers.json").write_text(json.dumps(timers_data)) (tmp_path / "memory" / "token-ledger.json").write_text(json.dumps(ledger_data)) sr = SessionResilience(workspace_root=str(tmp_path)) with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) result = sr.check_all_sessions() assert result["checked"] == 1 assert result["warnings"] == [] assert len(result["criticals"]) == 1 assert result["normals"] == 0 critical = result["criticals"][0] assert critical["task_id"] == task_id assert "usage_pct" in critical assert "event_path" in critical assert "summary_path" in critical assert "resume_triggered" in critical def test_check_session_returns_status(self, tmp_path: Path) -> None: """개별 세션 체크 결과 형식 확인""" from utils.session_resilience import SessionResilience sr = SessionResilience(workspace_root=str(tmp_path)) task_id = "task-500.1" task_info = {"task_id": task_id, "team_id": "dev2-team", "status": "running"} ledger_info = {"total_tokens": 100_000, "team_id": "dev2-team"} status = sr.check_session(task_id, task_info, ledger_info) assert status["task_id"] == task_id assert status["team_id"] == "dev2-team" assert status["total_tokens"] == 100_000 assert "usage_pct" in status assert status["level"] in ("normal", "warning", "critical") def test_handle_warning_creates_event_file(self, tmp_path: Path) -> None: """WARNING 이벤트 파일 생성 확인""" from utils.session_resilience import SessionResilience events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) sr = SessionResilience(workspace_root=str(tmp_path)) task_id = "task-600.1" team_id = "dev2-team" session_status = { "task_id": task_id, "team_id": team_id, "total_tokens": 150_000, "usage_pct": 75.0, "level": "warning", } result = sr.handle_warning(task_id, team_id, session_status) assert "event_path" in result assert result["level"] == "warning" event_path = Path(result["event_path"]) assert event_path.exists() assert f"session-warning-{task_id}" in event_path.name # JSON 형식 검증 event_data = json.loads(event_path.read_text()) assert event_data["level"] == "warning" assert event_data["task_id"] == task_id def test_handle_critical_creates_event_file(self, tmp_path: Path) -> None: """CRITICAL 이벤트 파일 생성 확인""" from utils.session_resilience import SessionResilience events_dir = tmp_path / "memory" / "events" sessions_dir = tmp_path / "memory" / "sessions" events_dir.mkdir(parents=True) sessions_dir.mkdir(parents=True) sr = SessionResilience(workspace_root=str(tmp_path)) task_id = "task-700.1" team_id = "dev2-team" session_status = { "task_id": task_id, "team_id": team_id, "total_tokens": 180_000, "usage_pct": 90.0, "level": "critical", } with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) result = sr.handle_critical(task_id, team_id, session_status) assert "event_path" in result assert result["level"] == "critical" event_path = Path(result["event_path"]) assert event_path.exists() assert f"session-critical-{task_id}" in event_path.name # JSON 형식 검증 event_data = json.loads(event_path.read_text()) assert event_data["level"] == "critical" assert event_data["task_id"] == task_id def test_handle_critical_saves_session_summary(self, tmp_path: Path) -> None: """CRITICAL 시 세션 요약 파일 저장 확인""" from utils.session_resilience import SessionResilience events_dir = tmp_path / "memory" / "events" sessions_dir = tmp_path / "memory" / "sessions" events_dir.mkdir(parents=True) sessions_dir.mkdir(parents=True) sr = SessionResilience(workspace_root=str(tmp_path)) task_id = "task-800.1" team_id = "dev2-team" session_status = { "task_id": task_id, "team_id": team_id, "total_tokens": 180_000, "usage_pct": 90.0, "level": "critical", } with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) result = sr.handle_critical(task_id, team_id, session_status) assert "summary_path" in result summary_path = Path(result["summary_path"]) assert summary_path.exists() assert f"summary-{task_id}" in summary_path.name assert summary_path.suffix == ".md" # 내용 검증 content = summary_path.read_text() assert task_id in content assert team_id in content def test_handle_critical_triggers_resume(self, tmp_path: Path) -> None: """CRITICAL 시 resume 트리거 (dispatch 호출 mock)""" from utils.session_resilience import SessionResilience events_dir = tmp_path / "memory" / "events" sessions_dir = tmp_path / "memory" / "sessions" events_dir.mkdir(parents=True) sessions_dir.mkdir(parents=True) sr = SessionResilience(workspace_root=str(tmp_path)) task_id = "task-900.1" team_id = "dev2-team" session_status = { "task_id": task_id, "team_id": team_id, "total_tokens": 180_000, "usage_pct": 90.0, "level": "critical", } with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) result = sr.handle_critical(task_id, team_id, session_status) # subprocess.run이 호출되었는지 확인 (dispatch.py --resume-from) mock_run.assert_called_once() call_args = mock_run.call_args cmd = call_args[0][0] assert "--resume-from" in cmd or any("--resume-from" in str(arg) for arg in cmd) assert result["resume_triggered"] is True def test_already_handled_session_not_re_triggered(self, tmp_path: Path) -> None: """이미 처리된 세션은 중복 트리거 방지""" from utils.session_resilience import SessionResilience events_dir = tmp_path / "memory" / "events" sessions_dir = tmp_path / "memory" / "sessions" events_dir.mkdir(parents=True) sessions_dir.mkdir(parents=True) task_id = "task-1000.1" team_id = "dev2-team" # 이미 critical 이벤트 파일이 존재하는 상황 시뮬레이션 existing_event = events_dir / f"session-critical-{task_id}-20240101T000000.json" existing_event.write_text(json.dumps({"level": "critical", "task_id": task_id})) # task-timers.json: 90% 사용으로 critical total_tokens = 180_000 timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(total_tokens, team_id)}) (tmp_path / "memory").mkdir(parents=True, exist_ok=True) (tmp_path / "memory" / "task-timers.json").write_text(json.dumps(timers_data)) (tmp_path / "memory" / "token-ledger.json").write_text(json.dumps(ledger_data)) sr = SessionResilience(workspace_root=str(tmp_path)) with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) result = sr.check_all_sessions() # 이미 처리된 세션이므로 subprocess.run 호출 없어야 함 mock_run.assert_not_called() # critical 리스트에는 없거나, resume_triggered=False if result["criticals"]: critical = result["criticals"][0] assert critical.get("resume_triggered") is False def test_get_bot_mapping(self, tmp_path: Path) -> None: """봇 ID ↔ 팀 매핑 확인""" from utils.session_resilience import BOT_TEAMS, SessionResilience # BOT_TEAMS 매핑 확인 assert "anu" in BOT_TEAMS assert "dev1-team" in BOT_TEAMS assert "dev2-team" in BOT_TEAMS assert BOT_TEAMS["dev2-team"] == "dev2" # SessionResilience 인스턴스에서도 접근 가능한지 확인 sr = SessionResilience(workspace_root=str(tmp_path)) assert hasattr(sr, "bot_teams") or BOT_TEAMS is not None def test_check_session_warning_level(self, tmp_path: Path) -> None: """check_session이 70% 이상에서 warning 레벨 반환""" from utils.session_resilience import SessionResilience sr = SessionResilience(workspace_root=str(tmp_path)) task_id = "task-1100.1" task_info = {"task_id": task_id, "team_id": "dev3-team", "status": "running"} # 75% = warning ledger_info = {"total_tokens": 150_000, "team_id": "dev3-team"} status = sr.check_session(task_id, task_info, ledger_info) assert status["level"] == "warning" assert status["usage_pct"] == pytest.approx(75.0) def test_check_session_critical_level(self, tmp_path: Path) -> None: """check_session이 85% 이상에서 critical 레벨 반환""" from utils.session_resilience import SessionResilience sr = SessionResilience(workspace_root=str(tmp_path)) task_id = "task-1200.1" task_info = {"task_id": task_id, "team_id": "dev4-team", "status": "running"} # 90% = critical ledger_info = {"total_tokens": 180_000, "team_id": "dev4-team"} status = sr.check_session(task_id, task_info, ledger_info) assert status["level"] == "critical" def test_check_all_sessions_multiple_tasks(self, tmp_path: Path) -> None: """여러 세션 동시 체크: normal, warning, critical 혼합""" from utils.session_resilience import SessionResilience # normal: 50% / warning: 75% / critical: 90% timers_data = _make_timers( { "task-n.1": _make_running_task("task-n.1", "dev1-team"), "task-w.1": _make_running_task("task-w.1", "dev2-team"), "task-c.1": _make_running_task("task-c.1", "dev3-team"), } ) ledger_data = _make_ledger( { "task-n.1": _make_ledger_entry(100_000, "dev1-team"), "task-w.1": _make_ledger_entry(150_000, "dev2-team"), "task-c.1": _make_ledger_entry(180_000, "dev3-team"), } ) (tmp_path / "memory").mkdir(parents=True) (tmp_path / "memory" / "task-timers.json").write_text(json.dumps(timers_data)) (tmp_path / "memory" / "token-ledger.json").write_text(json.dumps(ledger_data)) sr = SessionResilience(workspace_root=str(tmp_path)) with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) result = sr.check_all_sessions() assert result["checked"] == 3 assert result["normals"] == 1 assert len(result["warnings"]) == 1 assert len(result["criticals"]) == 1