""" 테스트: test_task_180_1.py 대상: 1. /home/jay/workspace/prompts/team_prompts.py :: _build_glm_prompt() 2. /home/jay/workspace/scripts/check-dev3.py (check_openclaw, check_paths, scan_event_files, calc_stats, --json / --quick CLI) 작성자: 아르고스 (테스터, dev1-team) """ import importlib import json import subprocess import sys import tempfile from pathlib import Path from unittest.mock import MagicMock, patch import pytest # ── 경로 설정 ────────────────────────────────────────────────────────────────── WORKSPACE = "/home/jay/workspace" sys.path.insert(0, WORKSPACE) # ── 모듈 임포트 ──────────────────────────────────────────────────────────────── from prompts.team_prompts import _build_glm_prompt, TEAM_INFO, WORKSPACE_ROOT # check-dev3.py 는 하이픈이 포함되어 있으므로 importlib.util 로 로드 import importlib.util as _ilu _spec = _ilu.spec_from_file_location( "check_dev3", f"{WORKSPACE}/scripts/check-dev3.py" ) check_dev3 = _ilu.module_from_spec(_spec) _spec.loader.exec_module(check_dev3) # ═══════════════════════════════════════════════════════════════════════════════ # 헬퍼 – _build_glm_prompt 기본 인자 조합 # ═══════════════════════════════════════════════════════════════════════════════ def _make_glm_prompt( task_id="task-180.1", task_desc="테스트 작업 설명", level="normal", project_id=None, chain_id=None, ) -> str: team = TEAM_INFO["dev3-team"] team_id = "dev3-team" timer_start = f"python3 {WORKSPACE_ROOT}/memory/task-timer.py start {task_id} --team {team_id} --desc \"{task_desc[:20]}\"" timer_end = f"python3 {WORKSPACE_ROOT}/memory/task-timer.py end {task_id}" report_path = f"{WORKSPACE_ROOT}/memory/reports/{task_id}.md" return _build_glm_prompt( team=team, team_id=team_id, task_id=task_id, task_desc=task_desc, level=level, timer_start=timer_start, timer_end=timer_end, report_path=report_path, project_id=project_id, chain_id=chain_id, ) # ═══════════════════════════════════════════════════════════════════════════════ # 1. _build_glm_prompt() 테스트 # ═══════════════════════════════════════════════════════════════════════════════ class TestBuildGlmPrompt: """_build_glm_prompt() 단위 테스트""" # ------------------------------------------------------------------ # 1-1. 기본 프롬프트 생성: task_id, task_desc 포함 여부 # ------------------------------------------------------------------ def test_contains_task_id(self): """프롬프트에 task_id 가 포함돼야 한다.""" prompt = _make_glm_prompt(task_id="task-180.1", task_desc="API 서버 구축") assert "task-180.1" in prompt, "task_id 가 프롬프트에 없음" def test_contains_task_desc(self): """프롬프트에 task_desc 가 포함돼야 한다.""" desc = "API 서버 구축 및 테스트" prompt = _make_glm_prompt(task_desc=desc) assert desc in prompt, "task_desc 가 프롬프트에 없음" # ------------------------------------------------------------------ # 1-2. 로깅 관련 지시: log_dir, log_file 경로 # ------------------------------------------------------------------ def test_contains_log_dir(self): """프롬프트에 log_dir 경로가 포함돼야 한다.""" task_id = "task-180.1" prompt = _make_glm_prompt(task_id=task_id) log_dir = f"{WORKSPACE_ROOT}/teams/dev3/logs" assert log_dir in prompt, f"log_dir ({log_dir}) 가 프롬프트에 없음" def test_contains_log_file(self): """프롬프트에 log_file 경로가 포함돼야 한다.""" task_id = "task-180.1" prompt = _make_glm_prompt(task_id=task_id) log_file = f"{WORKSPACE_ROOT}/teams/dev3/logs/{task_id}.log" assert log_file in prompt, f"log_file ({log_file}) 가 프롬프트에 없음" # ------------------------------------------------------------------ # 1-3. 재시도 메커니즘 지시: "1차 타임아웃", "재시도" # ------------------------------------------------------------------ def test_contains_timeout_retry_keywords(self): """프롬프트에 '1차 타임아웃' 과 '재시도' 키워드가 포함돼야 한다.""" prompt = _make_glm_prompt() assert "1차 타임아웃" in prompt, "'1차 타임아웃' 키워드 없음" assert "재시도" in prompt, "'재시도' 키워드 없음" # ------------------------------------------------------------------ # 1-4. 실패 통보 지시: failed 파일, cokacdir # ------------------------------------------------------------------ def test_contains_failed_file_and_cokacdir(self): """프롬프트에 .failed 파일 경로와 cokacdir 명령이 포함돼야 한다.""" task_id = "task-180.1" prompt = _make_glm_prompt(task_id=task_id) failed_file = f"{WORKSPACE_ROOT}/memory/events/{task_id}.failed" assert failed_file in prompt, f"failed_file ({failed_file}) 경로 없음" assert "cokacdir" in prompt, "cokacdir 명령어 없음" # ------------------------------------------------------------------ # 1-5. GLM-5 에러 핸들링: "오류 발생하면", "mkdir -p" # ------------------------------------------------------------------ def test_contains_glm_error_handling(self): """GLM 메시지에 오류 핸들링 및 mkdir -p 지시가 포함돼야 한다.""" prompt = _make_glm_prompt() assert "오류" in prompt, "오류 관련 문구 없음" assert "mkdir -p" in prompt, "mkdir -p 지시 없음" # ------------------------------------------------------------------ # 1-6. done 파일 경로: teams/dev3/{task_id}.done # ------------------------------------------------------------------ def test_done_file_path_correct(self): """done 파일 경로가 teams/dev3/{task_id}.done 형식이어야 한다.""" task_id = "task-180.1" prompt = _make_glm_prompt(task_id=task_id) expected_done = f"{WORKSPACE_ROOT}/teams/dev3/{task_id}.done" assert expected_done in prompt, f"done 파일 경로 ({expected_done}) 없음" # ------------------------------------------------------------------ # 1-7. project_id 가 있을 때 프로젝트 격리 규칙 포함 # ------------------------------------------------------------------ def test_project_isolation_with_project_id(self): """project_id 가 주어지면 프로젝트 격리 규칙이 포함돼야 한다.""" project_id = "proj-alpha" prompt = _make_glm_prompt(project_id=project_id) assert "프로젝트 격리" in prompt, "프로젝트 격리 문구 없음" assert f"{WORKSPACE_ROOT}/projects/{project_id}/" in prompt, "프로젝트 경로 없음" # ------------------------------------------------------------------ # 1-8. chain_id 가 있을 때 체인 완료 알림 포함 # ------------------------------------------------------------------ def test_chain_notification_with_chain_id(self): """chain_id 가 주어지면 체인 완료 알림 명령이 포함돼야 한다.""" chain_id = "chain-99" task_id = "task-180.1" prompt = _make_glm_prompt(task_id=task_id, chain_id=chain_id) assert "chain.py" in prompt, "chain.py 명령 없음" assert chain_id in prompt, f"chain_id ({chain_id}) 없음" assert "task-done" in prompt, "task-done 서브커맨드 없음" # ------------------------------------------------------------------ # 1-9. chain_id 가 None 일 때 체인 관련 내용 없음 # ------------------------------------------------------------------ def test_no_chain_notification_without_chain_id(self): """chain_id 가 None 이면 프롬프트에 체인 알림 명령이 없어야 한다.""" prompt = _make_glm_prompt(chain_id=None) # chain.py 는 완료 알림에만 사용됨 - chain_id 없으면 없어야 함 assert "task-done" not in prompt, "chain_id 없는데 task-done 명령이 포함됨" # ------------------------------------------------------------------ # 1-10. 타임아웃 시나리오: FOUND 변수, seq 1 120 # ------------------------------------------------------------------ def test_timeout_scenario_details(self): """프롬프트에 FOUND 변수와 seq 1 120 지시가 포함돼야 한다.""" prompt = _make_glm_prompt() assert "FOUND" in prompt, "FOUND 변수 없음" assert "seq 1 120" in prompt, "seq 1 120 없음" # ═══════════════════════════════════════════════════════════════════════════════ # 2. check-dev3.py 테스트 # ═══════════════════════════════════════════════════════════════════════════════ class TestCheckOpenclaw: """check_openclaw() 단위 테스트 (subprocess mock 사용)""" # ------------------------------------------------------------------ # 2-1. openclaw 가 존재하고 버전 반환이 정상인 경우 # ------------------------------------------------------------------ def test_openclaw_available(self): """which openclaw 성공 + --version 성공 → available=True, version 설정""" which_result = MagicMock() which_result.returncode = 0 which_result.stdout = "/usr/local/bin/openclaw\n" version_result = MagicMock() version_result.returncode = 0 version_result.stdout = "openclaw 1.2.3\n" version_result.stderr = "" with patch("subprocess.run", side_effect=[which_result, version_result]): result = check_dev3.check_openclaw() assert result["available"] is True assert result["path"] == "/usr/local/bin/openclaw" assert result["version"] == "openclaw 1.2.3" assert result["error"] is None # ------------------------------------------------------------------ # 2-2. openclaw 가 존재하지 않는 경우 # ------------------------------------------------------------------ def test_openclaw_not_found(self): """which openclaw 실패(returncode!=0) → available=False""" which_result = MagicMock() which_result.returncode = 1 which_result.stdout = "" with patch("subprocess.run", return_value=which_result): result = check_dev3.check_openclaw() assert result["available"] is False assert result["path"] is None # ------------------------------------------------------------------ # 2-3. subprocess 자체가 예외를 던지는 경우 # ------------------------------------------------------------------ def test_openclaw_subprocess_exception(self): """subprocess.run 이 FileNotFoundError 를 던지면 error 필드에 기록""" with patch("subprocess.run", side_effect=FileNotFoundError("which not found")): result = check_dev3.check_openclaw() assert result["available"] is False assert result["error"] is not None assert "which not found" in result["error"] # ------------------------------------------------------------------ # 2-4. which 는 성공했지만 --version 이 실패하는 경우 # ------------------------------------------------------------------ def test_openclaw_version_fails(self): """which 성공 + --version returncode!=0 → available=True, version 은 stderr""" which_result = MagicMock() which_result.returncode = 0 which_result.stdout = "/usr/bin/openclaw\n" version_result = MagicMock() version_result.returncode = 1 version_result.stdout = "" version_result.stderr = "unknown option" with patch("subprocess.run", side_effect=[which_result, version_result]): result = check_dev3.check_openclaw() assert result["available"] is True assert result["version"] == "unknown option" # ------------------------------------------------------------------ # 2-5. which 성공 + --version 타임아웃 # ------------------------------------------------------------------ def test_openclaw_version_timeout(self): """which 성공 + --version TimeoutExpired → available=True, error 설정""" which_result = MagicMock() which_result.returncode = 0 which_result.stdout = "/usr/bin/openclaw\n" def side_effect(cmd, **kwargs): if "which" in cmd: return which_result raise subprocess.TimeoutExpired(cmd, 5) with patch("subprocess.run", side_effect=side_effect): result = check_dev3.check_openclaw() assert result["available"] is True assert result["error"] is not None class TestCheckPaths: """check_paths() 테스트 (실제 환경)""" def test_returns_list_of_dicts(self): """결과가 리스트이고 각 항목에 필수 키가 있어야 한다.""" checks = check_dev3.check_paths() assert isinstance(checks, list) assert len(checks) > 0 for item in checks: assert "label" in item assert "path" in item assert "exists" in item assert "suggestion" in item def test_dev3_dir_check_present(self): """dev3 작업 디렉토리 항목이 포함돼야 한다.""" checks = check_dev3.check_paths() labels = [c["label"] for c in checks] assert "dev3 작업 디렉토리" in labels def test_events_dir_check_present(self): """events 디렉토리 항목이 포함돼야 한다.""" checks = check_dev3.check_paths() labels = [c["label"] for c in checks] assert "events 디렉토리" in labels def test_exists_is_boolean(self): """exists 필드는 bool 타입이어야 한다.""" checks = check_dev3.check_paths() for item in checks: assert isinstance(item["exists"], bool) def test_suggestion_for_missing_logs_dir(self): """로그 디렉토리가 없을 경우 suggestion 에 mkdir 명령이 포함돼야 한다.""" checks = check_dev3.check_paths() log_check = next((c for c in checks if c["label"] == "dev3 로그 디렉토리"), None) assert log_check is not None, "dev3 로그 디렉토리 항목 없음" if not log_check["exists"]: assert log_check["suggestion"] is not None assert "mkdir" in log_check["suggestion"] class TestScanEventFiles: """scan_event_files() 테스트 (실제 환경)""" def test_returns_list(self): """반환값은 리스트여야 한다.""" records = check_dev3.scan_event_files() assert isinstance(records, list) def test_each_record_has_required_keys(self): """각 레코드에 task_id, status, mtime, mtime_str, has_report 키가 있어야 한다.""" records = check_dev3.scan_event_files() required = {"task_id", "status", "mtime", "mtime_str", "has_report"} for r in records: missing = required - set(r.keys()) assert not missing, f"레코드에 키 누락: {missing}" def test_status_values_valid(self): """status 값은 'done' 또는 'failed' 여야 한다.""" records = check_dev3.scan_event_files() for r in records: assert r["status"] in ("done", "failed"), f"잘못된 status: {r['status']}" def test_no_duplicate_task_ids(self): """반환 결과에 중복 task_id 가 없어야 한다.""" records = check_dev3.scan_event_files() task_ids = [r["task_id"] for r in records] assert len(task_ids) == len(set(task_ids)), "중복 task_id 존재" def test_sorted_by_mtime_desc(self): """결과는 mtime 내림차순 정렬이어야 한다.""" records = check_dev3.scan_event_files() if len(records) >= 2: mtimes = [r["mtime"] for r in records] assert mtimes == sorted(mtimes, reverse=True), "mtime 내림차순 정렬 아님" def test_returns_empty_when_events_dir_missing(self): """EVENTS_DIR 이 없으면 빈 리스트를 반환해야 한다.""" with patch.object(check_dev3, "EVENTS_DIR", Path("/nonexistent/path/xyz")): records = check_dev3.scan_event_files() assert records == [] class TestCalcStats: """calc_stats() 단위 테스트""" def test_all_done(self): """전부 done 이면 success=total, failed=0, rate=100""" history = [{"status": "done"}, {"status": "done"}, {"status": "done"}] stats = check_dev3.calc_stats(history) assert stats["total"] == 3 assert stats["success"] == 3 assert stats["failed"] == 0 assert stats["rate"] == 100 def test_all_failed(self): """전부 failed 이면 success=0, rate=0""" history = [{"status": "failed"}, {"status": "failed"}] stats = check_dev3.calc_stats(history) assert stats["total"] == 2 assert stats["success"] == 0 assert stats["failed"] == 2 assert stats["rate"] == 0 def test_mixed(self): """2 done / 2 failed → rate=50""" history = [ {"status": "done"}, {"status": "failed"}, {"status": "done"}, {"status": "failed"}, ] stats = check_dev3.calc_stats(history) assert stats["total"] == 4 assert stats["success"] == 2 assert stats["failed"] == 2 assert stats["rate"] == 50 def test_empty_history(self): """빈 이력이면 rate=0, 모든 카운트 0""" stats = check_dev3.calc_stats([]) assert stats["total"] == 0 assert stats["success"] == 0 assert stats["failed"] == 0 assert stats["rate"] == 0 def test_single_done(self): """1건 done → rate=100""" stats = check_dev3.calc_stats([{"status": "done"}]) assert stats["rate"] == 100 def test_rate_truncates_not_rounds(self): """1/3 → int(33.3) = 33 (반올림 아닌 절사)""" history = [{"status": "done"}, {"status": "failed"}, {"status": "failed"}] stats = check_dev3.calc_stats(history) assert stats["rate"] == 33 def test_stats_keys_present(self): """반환 dict 에 total, success, failed, rate 키 존재""" stats = check_dev3.calc_stats([]) for key in ("total", "success", "failed", "rate"): assert key in stats, f"'{key}' 키 없음" class TestCliJson: """--json 플래그: 유효한 JSON 출력 검증""" def test_json_output_is_valid_json(self): """--json 모드로 실행하면 stdout 이 파싱 가능한 JSON 이어야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--json"], capture_output=True, text=True, timeout=30, ) assert proc.returncode == 0, f"exit code: {proc.returncode}\nstderr: {proc.stderr}" try: data = json.loads(proc.stdout) except json.JSONDecodeError as e: pytest.fail(f"JSON 파싱 실패: {e}\n출력: {proc.stdout[:500]}") # 최소 구조 확인 assert "openclaw" in data assert "paths" in data assert "history" in data assert "stats" in data assert "generated_at" in data def test_json_stats_structure(self): """JSON 출력의 stats 에 total/success/failed/rate 키가 있어야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--json"], capture_output=True, text=True, timeout=30, ) data = json.loads(proc.stdout) stats = data["stats"] for key in ("total", "success", "failed", "rate"): assert key in stats, f"stats 에 '{key}' 없음" def test_json_paths_is_list(self): """JSON 출력의 paths 는 리스트여야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--json"], capture_output=True, text=True, timeout=30, ) data = json.loads(proc.stdout) assert isinstance(data["paths"], list) def test_json_openclaw_has_available_key(self): """JSON 출력의 openclaw 에 available 키가 있어야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--json"], capture_output=True, text=True, timeout=30, ) data = json.loads(proc.stdout) assert "available" in data["openclaw"] class TestCliQuick: """--quick 플래그: 환경 점검만 출력 검증""" def test_quick_exits_zero(self): """--quick 모드는 exit code 0 이어야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--quick"], capture_output=True, text=True, timeout=30, ) assert proc.returncode == 0, f"exit code: {proc.returncode}\nstderr: {proc.stderr}" def test_quick_output_contains_env_check(self): """--quick 출력에 '환경 점검' 섹션이 포함돼야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--quick"], capture_output=True, text=True, timeout=30, ) assert "환경 점검" in proc.stdout, "'환경 점검' 문자열 없음" def test_quick_output_no_history(self): """--quick 출력에 '최근 작업 이력' 섹션이 없어야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--quick"], capture_output=True, text=True, timeout=30, ) assert "최근 작업 이력" not in proc.stdout, "--quick 인데 이력 섹션이 출력됨" def test_quick_output_no_stats(self): """--quick 출력에 통계(성공률) 섹션이 없어야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--quick"], capture_output=True, text=True, timeout=30, ) assert "성공률" not in proc.stdout, "--quick 인데 통계 섹션이 출력됨" def test_quick_output_contains_health_check_title(self): """--quick 출력 첫 줄에 헬스체크 제목이 포함돼야 한다.""" proc = subprocess.run( [sys.executable, f"{WORKSPACE}/scripts/check-dev3.py", "--quick"], capture_output=True, text=True, timeout=30, ) assert "Dev3 Team Health Check" in proc.stdout