""" test_integration.py 모듈 간 연동 통합 테스트 (헤임달 작성) 시나리오: 1. dispatch 전체 흐름: generate_task_id() → build_prompt() → get_dispatch_time() 2. task-timer 전체 라이프사이클: start → list → end → list 3. 로거 + 에러 추적 연동: logger.py → error_tracker.py → get_recent_errors() 4. team_prompts 전팀 프롬프트 생성: dev1/dev2/dev3 모두 검증 5. run_tests.py 실행 검증: subprocess → JSON 파싱 → passed == total 격리 원칙: tmp_path로 파일시스템 완전 격리 """ import importlib.util import json import os import subprocess import sys from datetime import datetime from pathlib import Path 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)) # --------------------------------------------------------------------------- # 공통 헬퍼: dispatch 모듈 격리 로드 # --------------------------------------------------------------------------- def _load_dispatch_isolated(tmp_path: Path): """dispatch 모듈을 tmp_path를 WORKSPACE로 설정하여 로드""" (tmp_path / "memory").mkdir(parents=True, exist_ok=True) (tmp_path / "memory" / "tasks").mkdir(parents=True, exist_ok=True) # dispatch 의존 모듈 미리 로드 import prompts.team_prompts # noqa: F401 # dispatch 모듈 재로드 for mod_name in list(sys.modules.keys()): if mod_name == "dispatch": del sys.modules[mod_name] import dispatch as _dispatch _dispatch.WORKSPACE = tmp_path return _dispatch # --------------------------------------------------------------------------- # 공통 헬퍼: task-timer 모듈 로드 # --------------------------------------------------------------------------- _TIMER_PATH = WORKSPACE / "memory" / "task-timer.py" def _load_task_timer(): """하이픈 포함 파일명을 importlib.util로 로드""" spec = importlib.util.spec_from_file_location("task_timer_integ", _TIMER_PATH) assert spec is not None mod = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(mod) return mod _task_timer_mod = _load_task_timer() TaskTimer = _task_timer_mod.TaskTimer # --------------------------------------------------------------------------- # 공통 헬퍼: team_prompts 격리 build_prompt # --------------------------------------------------------------------------- def _make_safe_build(tmp_path: Path): """tmp_path에 task 파일을 저장하는 격리된 build_prompt 생성""" import prompts.team_prompts as tp def _safe_build(team_id, task_id, task_desc, level="normal", project_id=None, chain_id=None, task_type="coding"): task_dir = tmp_path / "memory" / "tasks" task_dir.mkdir(parents=True, exist_ok=True) task_file = task_dir / f"{task_id}.md" task_file.write_text(task_desc, encoding="utf-8") team = tp.TEAM_INFO.get(team_id) if not team: raise ValueError(f"알 수 없는 팀 ID: {team_id}") short_desc = task_desc[:20] timer_start = f'python3 ... start {task_id} --team {team_id} --desc "{short_desc}"' timer_end = f"python3 ... end {task_id}" report_path = str(tmp_path / "memory" / "reports" / f"{task_id}.md") if team["type"] == "direct": prompt = tp._build_direct_prompt( team, team_id, task_id, str(task_file), level, timer_start, timer_end, report_path, project_id=project_id, ) else: prompt = tp._build_glm_prompt( team, team_id, task_id, task_desc, level, timer_start, timer_end, report_path, project_id=project_id, ) if task_type == "coding": prompt += tp._build_verification_section(level) if level == "critical": prompt = "**[CRITICAL] 이 작업은 중요도 critical입니다. 품질 우선으로 신중하게 작업하세요.**\n\n" + prompt elif level == "security": prompt = "**[SECURITY] 이 작업은 보안 중요 작업입니다. 보안 최우선으로 작업하세요.**\n\n" + prompt return prompt return _safe_build # =========================================================================== # 시나리오 1: dispatch 전체 흐름 # =========================================================================== class TestIntegrationDispatchFlow: """generate_task_id() → build_prompt() → get_dispatch_time() 순차 연동 검증""" def test_task_id_passed_to_build_prompt(self, tmp_path, monkeypatch): """generate_task_id()로 생성한 ID가 build_prompt() 출력에 포함되는지 검증""" dispatch_mod = _load_dispatch_isolated(tmp_path) # _build_team_prompt를 격리된 버전으로 교체 safe_build = _make_safe_build(tmp_path) monkeypatch.setattr(dispatch_mod, "_build_team_prompt", safe_build) # Step 1: generate_task_id() → task-1 task_id = dispatch_mod.generate_task_id() assert task_id == "task-1" # Step 2: build_prompt() 호출 — Step 1 결과를 입력으로 사용 prompt = dispatch_mod.build_prompt("dev2-team", "통합 테스트 작업", task_id) assert isinstance(prompt, str) assert task_id in prompt # 연동 검증: task_id가 프롬프트에 포함 def test_generate_id_recorded_before_build_prompt(self, tmp_path, monkeypatch): """generate_task_id() 직후 ID가 timer 파일에 기록되어 있어야 함""" dispatch_mod = _load_dispatch_isolated(tmp_path) safe_build = _make_safe_build(tmp_path) monkeypatch.setattr(dispatch_mod, "_build_team_prompt", safe_build) task_id = dispatch_mod.generate_task_id() # build_prompt 전에 이미 파일에 기록되어 있어야 함 timer_file = tmp_path / "memory" / "task-timers.json" assert timer_file.exists() data = json.loads(timer_file.read_text()) assert task_id in data["tasks"] # build_prompt 호출 후에도 파일 유지 dispatch_mod.build_prompt("dev1-team", "빌드 프롬프트 테스트", task_id) data_after = json.loads(timer_file.read_text()) assert task_id in data_after["tasks"] def test_get_dispatch_time_returns_future(self, tmp_path): """get_dispatch_time()이 현재 시각보다 미래의 시각을 반환하는지 검증""" dispatch_mod = _load_dispatch_isolated(tmp_path) before = datetime.now() dispatch_time = dispatch_mod.get_dispatch_time(30) parsed = datetime.strptime(dispatch_time, "%Y-%m-%d %H:%M:%S") assert parsed > before def test_full_flow_sequential(self, tmp_path, monkeypatch): """세 함수의 순차 호출 결과가 전체적으로 일관성 있는지 검증""" dispatch_mod = _load_dispatch_isolated(tmp_path) safe_build = _make_safe_build(tmp_path) monkeypatch.setattr(dispatch_mod, "_build_team_prompt", safe_build) # Step 1 task_id = dispatch_mod.generate_task_id() assert task_id.startswith("task-") # Step 2 prompt = dispatch_mod.build_prompt("dev2-team", "전체 흐름 검증 작업", task_id) assert task_id in prompt assert "오딘" in prompt # dev2-team 팀장 # Step 3 dispatch_time_str = dispatch_mod.get_dispatch_time(10) dispatch_time = datetime.strptime(dispatch_time_str, "%Y-%m-%d %H:%M:%S") assert dispatch_time > datetime.now() # 종합: 세 단계 모두 정상 완료 assert task_id and prompt and dispatch_time_str # =========================================================================== # 시나리오 2: task-timer 전체 라이프사이클 # =========================================================================== class TestIntegrationTaskTimerLifecycle: """start → list(running 확인) → end → list(completed 확인) 전체 흐름 검증""" @pytest.fixture() def timer(self, tmp_path): (tmp_path / "memory").mkdir(parents=True, exist_ok=True) return TaskTimer(workspace_path=str(tmp_path)) def test_full_lifecycle(self, timer): """start → list(running) → end → list(completed) 전체 흐름""" task_id = "task-999.2" # Step 1: start start_result = timer.start_task(task_id, team_id="dev2-team", description="라이프사이클 통합 테스트") assert start_result["status"] == "started" assert start_result["task_id"] == task_id # Step 2: list — running 확인 running_list = timer.list_tasks(status="running") assert running_list["total"] == 1 assert running_list["tasks"][0]["task_id"] == task_id # 완료 상태는 아직 없어야 함 completed_before = timer.list_tasks(status="completed") assert completed_before["total"] == 0 # Step 3: end end_result = timer.end_task(task_id) assert end_result["status"] == "completed" assert end_result["task_id"] == task_id assert "duration_seconds" in end_result # Step 4: list — completed 확인 completed_list = timer.list_tasks(status="completed") assert completed_list["total"] == 1 assert completed_list["tasks"][0]["task_id"] == task_id # running 목록은 비어야 함 running_after = timer.list_tasks(status="running") assert running_after["total"] == 0 def test_multiple_tasks_lifecycle(self, timer): """여러 작업 동시 진행 시 상태 구분이 올바르게 동작하는지 검증""" timer.start_task("task-100.1", description="작업 A") timer.start_task("task-200.1", description="작업 B") timer.start_task("task-300.1", description="작업 C") # 3개 모두 running assert timer.list_tasks(status="running")["total"] == 3 # task-100.1만 완료 timer.end_task("task-100.1") assert timer.list_tasks(status="running")["total"] == 2 assert timer.list_tasks(status="completed")["total"] == 1 # task-200.1도 완료 timer.end_task("task-200.1") assert timer.list_tasks(status="running")["total"] == 1 assert timer.list_tasks(status="completed")["total"] == 2 # 전체 조회 all_tasks = timer.list_tasks() assert all_tasks["total"] == 3 # =========================================================================== # 시나리오 3: 로거 + 에러 추적 연동 # =========================================================================== class TestIntegrationLoggerErrorTracker: """logger.py 로그 기록 → error_tracker.py 에러 기록 → get_recent_errors() 조회 검증""" def test_logger_writes_file_and_error_tracker_records(self, tmp_path, monkeypatch): """로그 파일 생성 + 에러 JSONL 기록 + 조회 연동 검증""" import utils.error_tracker as et_mod import utils.logger as logger_mod # 경로 격리 log_dir = tmp_path / "logs" log_dir.mkdir(parents=True, exist_ok=True) fake_log_file = log_dir / "app.log" fake_errors_file = log_dir / "errors.jsonl" monkeypatch.setattr(logger_mod, "LOG_FILE", fake_log_file) monkeypatch.setattr(et_mod, "ERRORS_FILE", fake_errors_file) # Step 1: 로거로 로그 기록 (유니크한 이름으로 신규 로거 생성) unique_logger_name = f"integ_test_{tmp_path.name}" logger = logger_mod.get_logger(unique_logger_name) logger.info("통합 테스트 로그 메시지") # 로그 파일이 생성되어야 함 assert fake_log_file.exists() # Step 2: 에러 추적기로 에러 기록 try: raise ValueError("통합 테스트용 에러 XYZ") except ValueError as e: et_mod.track_error("test_integration", e, tb_str="fake traceback") # 에러 파일이 생성되어야 함 assert fake_errors_file.exists() # Step 3: get_recent_errors()로 조회 — 연동 검증 errors = et_mod.get_recent_errors(10) assert len(errors) >= 1 last_error = errors[-1] assert last_error["module"] == "test_integration" assert last_error["error_type"] == "ValueError" assert "통합 테스트용 에러 XYZ" in last_error["message"] def test_multiple_errors_retrievable_in_order(self, tmp_path, monkeypatch): """여러 에러를 기록했을 때 순서대로 조회되는지 검증""" import utils.error_tracker as et_mod fake_errors_file = tmp_path / "errors.jsonl" monkeypatch.setattr(et_mod, "ERRORS_FILE", fake_errors_file) # 에러 3개 기록 for i in range(3): try: raise RuntimeError(f"에러 번호 {i}") except RuntimeError as e: et_mod.track_error(f"module_{i}", e, tb_str=f"tb_{i}") errors = et_mod.get_recent_errors(10) assert len(errors) == 3 # 모듈 이름 순서 확인 assert errors[0]["module"] == "module_0" assert errors[1]["module"] == "module_1" assert errors[2]["module"] == "module_2" def test_get_recent_errors_returns_empty_when_no_file(self, tmp_path, monkeypatch): """에러 파일이 없을 때 get_recent_errors()가 빈 리스트를 반환하는지 검증""" import utils.error_tracker as et_mod nonexistent_file = tmp_path / "nonexistent_errors.jsonl" monkeypatch.setattr(et_mod, "ERRORS_FILE", nonexistent_file) errors = et_mod.get_recent_errors(10) assert errors == [] # =========================================================================== # 시나리오 4: team_prompts 전팀 프롬프트 생성 # =========================================================================== class TestIntegrationTeamPrompts: """dev1/dev2/dev3 모든 팀에 대해 build_prompt() 호출 및 내용 검증""" @pytest.fixture() def safe_build(self, tmp_path): return _make_safe_build(tmp_path) def test_all_teams_prompts_generated_successfully(self, safe_build): """3개 팀 모두 프롬프트 생성 가능한지 검증""" teams = [ ("dev1-team", "task-1.1"), ("dev2-team", "task-2.1"), ("dev3-team", "task-3.1"), ] for team_id, task_id in teams: prompt = safe_build(team_id, task_id, f"{team_id} 통합 테스트 작업") assert isinstance(prompt, str), f"{team_id} 프롬프트가 문자열이 아님" assert len(prompt) > 0, f"{team_id} 프롬프트가 비어 있음" assert task_id in prompt, f"{team_id} 프롬프트에 {task_id} 누락" def test_direct_teams_contain_member_names(self, safe_build): """direct 타입(dev1, dev2)은 팀원 이름이 프롬프트에 포함되어야 함""" # dev1-team prompt1 = safe_build("dev1-team", "task-1.1", "dev1 통합 테스트") assert "헤르메스" in prompt1 for member in ["불칸", "이리스", "아테나", "아르고스"]: assert member in prompt1, f"dev1-team 프롬프트에 '{member}' 누락" # dev2-team prompt2 = safe_build("dev2-team", "task-2.1", "dev2 통합 테스트") assert "오딘" in prompt2 for member in ["토르", "프레이야", "미미르", "헤임달"]: assert member in prompt2, f"dev2-team 프롬프트에 '{member}' 누락" def test_glm_team_contains_openclaw_keyword(self, safe_build): """glm 타입(dev8)은 openclaw 키워드가 프롬프트에 포함되어야 함""" prompt = safe_build("dev8-team", "task-8.1", "dev8 GLM 통합 테스트") assert "openclaw" in prompt.lower() or "GLM" in prompt, "dev8-team 프롬프트에 openclaw/GLM 키워드 누락" assert "dev8-team" in prompt def test_all_prompts_contain_task_id(self, safe_build): """생성된 모든 팀 프롬프트에 해당 task_id가 포함되어야 함""" for i, team_id in enumerate(["dev1-team", "dev2-team", "dev3-team"], 1): task_id = f"task-{i}.1" prompt = safe_build(team_id, task_id, "task_id 포함 검증") assert task_id in prompt, f"{team_id} 프롬프트에 {task_id} 누락" # =========================================================================== # 시나리오 5: run_tests.py 실행 검증 # =========================================================================== class TestIntegrationRunTests: """subprocess로 run_tests.py 실행 후 JSON 파싱 및 결과 검증""" def test_run_tests_produces_valid_json(self): """run_tests.py가 실행되고 유효한 JSON을 stdout에 출력하는지 확인""" run_tests_path = WORKSPACE / "tests" / "run_tests.py" proc = subprocess.run( [sys.executable, str(run_tests_path)], capture_output=True, text=True, timeout=60, ) assert proc.stdout, "run_tests.py가 stdout에 아무것도 출력하지 않았습니다" result = json.loads(proc.stdout) assert isinstance(result, dict) def test_run_tests_output_has_required_keys(self): """JSON 출력에 total, passed, failed 키가 존재하는지 확인""" run_tests_path = WORKSPACE / "tests" / "run_tests.py" proc = subprocess.run( [sys.executable, str(run_tests_path)], capture_output=True, text=True, timeout=60, ) result = json.loads(proc.stdout) assert "total" in result, "JSON에 'total' 키 누락" assert "passed" in result, "JSON에 'passed' 키 누락" assert "failed" in result, "JSON에 'failed' 키 누락" def test_run_tests_all_passed(self): """passed == total (0 failures)""" run_tests_path = WORKSPACE / "tests" / "run_tests.py" proc = subprocess.run( [sys.executable, str(run_tests_path)], capture_output=True, text=True, timeout=60, ) result = json.loads(proc.stdout) assert result["passed"] == result["total"], ( f"일부 run_tests 실패: total={result['total']}, " f"passed={result['passed']}, failed={result['failed']}, " f"details={result.get('results', [])}" ) def test_run_tests_results_list_exists(self): """JSON 출력에 results 배열이 존재하고 total과 개수가 일치하는지 확인""" run_tests_path = WORKSPACE / "tests" / "run_tests.py" proc = subprocess.run( [sys.executable, str(run_tests_path)], capture_output=True, text=True, timeout=60, ) result = json.loads(proc.stdout) assert "results" in result assert isinstance(result["results"], list) assert len(result["results"]) == result["total"]