""" tests/test_session_watchdog.py session_watchdog.py 단위 테스트 (TDD) 테스트 항목: TestWatchdogCheck - test_check_returns_result: --check 실행 시 check_all_sessions() 결과 반환 - test_check_prints_json_output: --check 실행 시 JSON 출력 - test_check_with_no_running_sessions: 세션 없을 때 정상 동작 - test_check_detects_warning: warning 세션 감지 - test_check_detects_critical: critical 세션 감지 TestWatchdogStatus - test_status_returns_session_list: --status 실행 시 세션 목록 반환 - test_status_output_format: 출력 형식 확인 (JSON) TestWatchdogIntegration - test_main_function_check_mode: main() 함수 --check 모드 실행 - test_main_function_status_mode: main() 함수 --status 모드 실행 - test_exit_code_success: 정상 종료 시 exit code 0 - test_exit_code_on_critical: critical 감지 시 exit code 1 """ from __future__ import annotations import json import os import sys from io import StringIO 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, } def _setup_workspace(tmp_path: Path, timers_data: dict, ledger_data: dict) -> None: """tmp_path 아래 memory 디렉토리와 JSON 파일 생성.""" memory_dir = tmp_path / "memory" memory_dir.mkdir(parents=True, exist_ok=True) (memory_dir / "task-timers.json").write_text(json.dumps(timers_data), encoding="utf-8") (memory_dir / "token-ledger.json").write_text(json.dumps(ledger_data), encoding="utf-8") # --------------------------------------------------------------------------- # 1. TestWatchdogCheck # --------------------------------------------------------------------------- class TestWatchdogCheck: """run_check() 함수 테스트""" def test_check_returns_result(self, tmp_path: Path) -> None: """--check 실행 시 check_all_sessions() 결과 dict 반환.""" from scripts.session_watchdog import run_check _setup_workspace(tmp_path, _make_timers({}), _make_ledger({})) result = run_check(workspace_root=str(tmp_path)) assert isinstance(result, dict) assert "checked" in result assert "warnings" in result assert "criticals" in result assert "normals" in result assert "timestamp" in result def test_check_prints_json_output(self, tmp_path: Path, capsys: pytest.CaptureFixture) -> None: """--check 실행 시 stdout에 JSON 출력.""" from scripts.session_watchdog import run_check _setup_workspace(tmp_path, _make_timers({}), _make_ledger({})) # run_check 후 main()이 JSON 출력하는지는 main() 테스트에서 확인 # 여기서는 run_check 결과가 JSON 직렬화 가능한지 검증 result = run_check(workspace_root=str(tmp_path)) serialized = json.dumps(result, ensure_ascii=False) parsed = json.loads(serialized) assert parsed["checked"] == 0 def test_check_with_no_running_sessions(self, tmp_path: Path) -> None: """실행 중인 세션 없을 때 checked=0, 빈 리스트 반환.""" from scripts.session_watchdog import run_check timers_data = _make_timers( { "task-100.1": { **_make_running_task("task-100.1", "dev2-team"), "status": "completed", } } ) _setup_workspace(tmp_path, timers_data, _make_ledger({})) result = run_check(workspace_root=str(tmp_path)) assert result["checked"] == 0 assert result["warnings"] == [] assert result["criticals"] == [] assert result["normals"] == 0 def test_check_detects_warning(self, tmp_path: Path) -> None: """75% 사용 세션을 warning으로 감지.""" from scripts.session_watchdog import run_check task_id = "task-200.1" team_id = "dev2-team" # 150_000 / 200_000 = 75% → warning timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(150_000, team_id)}) _setup_workspace(tmp_path, timers_data, ledger_data) result = run_check(workspace_root=str(tmp_path)) assert result["checked"] == 1 assert len(result["warnings"]) == 1 assert result["criticals"] == [] assert result["warnings"][0]["task_id"] == task_id def test_check_detects_critical(self, tmp_path: Path) -> None: """90% 사용 세션을 critical로 감지.""" from scripts.session_watchdog import run_check task_id = "task-300.1" team_id = "dev2-team" # 180_000 / 200_000 = 90% → critical timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(180_000, team_id)}) _setup_workspace(tmp_path, timers_data, ledger_data) with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) result = run_check(workspace_root=str(tmp_path)) assert result["checked"] == 1 assert result["warnings"] == [] assert len(result["criticals"]) == 1 assert result["criticals"][0]["task_id"] == task_id # --------------------------------------------------------------------------- # 2. TestWatchdogStatus # --------------------------------------------------------------------------- class TestWatchdogStatus: """run_status() 함수 테스트""" def test_status_returns_session_list(self, tmp_path: Path) -> None: """--status 실행 시 'sessions' 키를 가진 dict 반환.""" from scripts.session_watchdog import run_status 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(100_000, team_id)}) _setup_workspace(tmp_path, timers_data, ledger_data) result = run_status(workspace_root=str(tmp_path)) assert isinstance(result, dict) assert "sessions" in result assert isinstance(result["sessions"], list) assert len(result["sessions"]) == 1 session = result["sessions"][0] assert session["task_id"] == task_id def test_status_output_format(self, tmp_path: Path) -> None: """status 결과가 JSON 직렬화 가능하고 필수 필드를 포함.""" from scripts.session_watchdog import run_status task_id = "task-500.1" team_id = "dev3-team" timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(50_000, team_id)}) _setup_workspace(tmp_path, timers_data, ledger_data) result = run_status(workspace_root=str(tmp_path)) # JSON 직렬화 가능 확인 serialized = json.dumps(result, ensure_ascii=False) parsed = json.loads(serialized) session = parsed["sessions"][0] assert "task_id" in session assert "team_id" in session assert "total_tokens" in session assert "usage_pct" in session assert "level" in session # --------------------------------------------------------------------------- # 3. TestWatchdogIntegration # --------------------------------------------------------------------------- class TestWatchdogIntegration: """main() 함수 통합 테스트""" def test_main_function_check_mode(self, tmp_path: Path, capsys: pytest.CaptureFixture) -> None: """main() --check 모드에서 JSON 출력하고 정상 종료.""" from scripts.session_watchdog import main _setup_workspace(tmp_path, _make_timers({}), _make_ledger({})) exit_code = main(["--check", "--workspace", str(tmp_path)]) captured = capsys.readouterr() assert exit_code == 0 # stdout에 JSON 출력 확인 output = json.loads(captured.out) assert "checked" in output assert "warnings" in output assert "criticals" in output def test_main_function_status_mode(self, tmp_path: Path, capsys: pytest.CaptureFixture) -> None: """main() --status 모드에서 세션 목록 JSON 출력.""" from scripts.session_watchdog import main _setup_workspace(tmp_path, _make_timers({}), _make_ledger({})) exit_code = main(["--status", "--workspace", str(tmp_path)]) captured = capsys.readouterr() assert exit_code == 0 output = json.loads(captured.out) assert "sessions" in output def test_exit_code_success(self, tmp_path: Path) -> None: """normal/warning만 있을 때 exit code 0.""" from scripts.session_watchdog import main task_id = "task-600.1" team_id = "dev2-team" # 50% → normal timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(100_000, team_id)}) _setup_workspace(tmp_path, timers_data, ledger_data) exit_code = main(["--check", "--workspace", str(tmp_path)]) assert exit_code == 0 def test_exit_code_on_critical(self, tmp_path: Path) -> None: """critical 세션 감지 시 exit code 1.""" from scripts.session_watchdog import main task_id = "task-700.1" team_id = "dev2-team" # 90% → critical timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(180_000, team_id)}) _setup_workspace(tmp_path, timers_data, ledger_data) with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) exit_code = main(["--check", "--workspace", str(tmp_path)]) assert exit_code == 1 def test_exit_code_warning_only_is_zero(self, tmp_path: Path) -> None: """warning만 있고 critical 없으면 exit code 0.""" from scripts.session_watchdog import main task_id = "task-800.1" team_id = "dev2-team" # 75% → warning timers_data = _make_timers({task_id: _make_running_task(task_id, team_id)}) ledger_data = _make_ledger({task_id: _make_ledger_entry(150_000, team_id)}) _setup_workspace(tmp_path, timers_data, ledger_data) exit_code = main(["--check", "--workspace", str(tmp_path)]) assert exit_code == 0 def test_status_mode_no_running_sessions(self, tmp_path: Path, capsys: pytest.CaptureFixture) -> None: """세션 없을 때 --status 모드 정상 동작.""" from scripts.session_watchdog import main _setup_workspace(tmp_path, _make_timers({}), _make_ledger({})) exit_code = main(["--status", "--workspace", str(tmp_path)]) captured = capsys.readouterr() assert exit_code == 0 output = json.loads(captured.out) assert output["sessions"] == [] def test_check_result_has_timestamp(self, tmp_path: Path) -> None: """check 결과에 timestamp 필드 포함.""" from scripts.session_watchdog import run_check _setup_workspace(tmp_path, _make_timers({}), _make_ledger({})) result = run_check(workspace_root=str(tmp_path)) assert "timestamp" in result assert result["timestamp"] != "" def test_run_check_normals_count(self, tmp_path: Path) -> None: """normal 세션 수가 정확히 반환됨.""" from scripts.session_watchdog import run_check # 50% × 3 → normal × 3 timers_data = _make_timers( { "task-n1.1": _make_running_task("task-n1.1", "dev1-team"), "task-n2.1": _make_running_task("task-n2.1", "dev2-team"), "task-n3.1": _make_running_task("task-n3.1", "dev3-team"), } ) ledger_data = _make_ledger( { "task-n1.1": _make_ledger_entry(100_000, "dev1-team"), "task-n2.1": _make_ledger_entry(100_000, "dev2-team"), "task-n3.1": _make_ledger_entry(100_000, "dev3-team"), } ) _setup_workspace(tmp_path, timers_data, ledger_data) result = run_check(workspace_root=str(tmp_path)) assert result["checked"] == 3 assert result["normals"] == 3 assert result["warnings"] == [] assert result["criticals"] == [] def test_run_status_returns_empty_when_no_files(self, tmp_path: Path) -> None: """timers/ledger 파일 없을 때 run_status가 빈 sessions 반환.""" from scripts.session_watchdog import run_status # 파일을 생성하지 않고 호출 result = run_status(workspace_root=str(tmp_path)) assert isinstance(result, dict) assert "sessions" in result assert result["sessions"] == []