"""dispatch.py _create_task_docs() 단위 테스트.""" import os import sys import types from pathlib import Path import pytest import yaml # --------------------------------------------------------------------------- # 헬퍼: dispatch 모듈을 격리된 WORKSPACE로 로드 (test_dispatch.py 패턴 참고) # --------------------------------------------------------------------------- def _load_dispatch_with_workspace(tmp_path: Path) -> types.ModuleType: """dispatch 모듈을 tmp_path를 WORKSPACE로 설정하여 로드한다.""" workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) # prompts 패키지를 먼저 로드 (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 # WORKSPACE를 tmp_path로 교체 _dispatch.WORKSPACE = tmp_path return _dispatch # --------------------------------------------------------------------------- # fixture: 격리된 dispatch 모듈 + 템플릿 파일 세팅 # --------------------------------------------------------------------------- REAL_WORKSPACE = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) REAL_TEMPLATE_DIR = REAL_WORKSPACE / "prompts" / "templates" / "task-docs" TEMPLATE_FILES = [ "plan.template.md", "context-notes.template.md", "checklist.template.md", ] @pytest.fixture() def dispatch_mod(tmp_path): """격리된 WORKSPACE를 사용하는 dispatch 모듈 반환 + 템플릿 파일 복사.""" # memory 디렉토리 구조 생성 (tmp_path / "memory").mkdir(parents=True, exist_ok=True) (tmp_path / "memory" / "tasks").mkdir(parents=True, exist_ok=True) # 템플릿 디렉토리 생성 및 실제 템플릿 복사 tmpl_dir = tmp_path / "prompts" / "templates" / "task-docs" tmpl_dir.mkdir(parents=True, exist_ok=True) for fname in TEMPLATE_FILES: src = REAL_TEMPLATE_DIR / fname if src.exists(): (tmpl_dir / fname).write_text(src.read_text(encoding="utf-8"), encoding="utf-8") mod = _load_dispatch_with_workspace(tmp_path) return mod # --------------------------------------------------------------------------- # 테스트 클래스 # --------------------------------------------------------------------------- class TestCreateTaskDocs: """_create_task_docs 함수 테스트""" # ------------------------------------------------------------------ # 테스트 1: level=3(critical)에서 디렉토리 + 파일 3개 생성 확인 # ------------------------------------------------------------------ def test_create_docs_for_critical_level(self, dispatch_mod, tmp_path): """Lv.3(critical)에서 3문서 디렉토리 + 파일 3개 생성""" task_id = "task-1872" result = dispatch_mod._create_task_docs(task_id, level=3) # 반환값이 Path 이어야 함 assert result is not None, "_create_task_docs가 None을 반환함" assert isinstance(result, Path) # 디렉토리가 생성되어야 함 expected_dir = tmp_path / "memory" / "plans" / "tasks" / task_id assert expected_dir.exists(), f"디렉토리가 생성되지 않음: {expected_dir}" assert expected_dir.is_dir() # 파일 3개가 존재해야 함 expected_files = ["plan.md", "context-notes.md", "checklist.md"] for fname in expected_files: fpath = expected_dir / fname assert fpath.exists(), f"파일이 생성되지 않음: {fpath}" # ------------------------------------------------------------------ # 테스트 2: 3파일 존재 + YAML frontmatter 유효성 확인 # ------------------------------------------------------------------ def test_yaml_frontmatter_valid(self, dispatch_mod, tmp_path): """생성된 파일의 YAML frontmatter에 task_id, type, scope, status 필드 존재""" task_id = "task-1872.1" dispatch_mod._create_task_docs(task_id, level=3) expected_dir = tmp_path / "memory" / "plans" / "tasks" / task_id expected_files = ["plan.md", "context-notes.md", "checklist.md"] for fname in expected_files: fpath = expected_dir / fname assert fpath.exists(), f"파일 없음: {fpath}" content = fpath.read_text(encoding="utf-8") # YAML frontmatter 추출: --- 와 --- 사이 assert content.startswith("---"), f"{fname}: YAML frontmatter 없음" end_idx = content.index("---", 3) yaml_text = content[3:end_idx].strip() data = yaml.safe_load(yaml_text) assert "task_id" in data, f"{fname}: task_id 필드 없음" assert "type" in data, f"{fname}: type 필드 없음" assert "scope" in data, f"{fname}: scope 필드 없음" assert "status" in data, f"{fname}: status 필드 없음" # task_id 치환 확인 assert data["task_id"] == task_id, f"{fname}: task_id 치환 실패 (기대: {task_id}, 실제: {data['task_id']})" # {date} 치환 확인 — 플레이스홀더가 남아있으면 안 됨 assert "{date}" not in content, f"{fname}: {{date}} 치환 미완료" assert "{task_id}" not in content, f"{fname}: {{task_id}} 치환 미완료" # ------------------------------------------------------------------ # 테스트 3: level=2(normal)에서 None 반환 / 디렉토리 미생성 # ------------------------------------------------------------------ def test_no_docs_for_normal_level(self, dispatch_mod, tmp_path): """Lv.2(normal)에서는 함수가 None 반환하고 디렉토리 미생성 dispatch() 내부 로직: _docs_level >= 3 조건에서만 _create_task_docs 호출. 함수 자체는 level 인자를 받지만, 실제 레벨 판단은 dispatch()에서 수행. level=2로 직접 호출 시에도 함수는 정상 동작(디렉토리 생성)할 수 있으나, dispatch() 흐름에서는 level<3이면 호출되지 않음을 검증한다. """ task_id = "task-1873" # dispatch 흐름에서 level<3 조건 재현: 직접 호출하지 않고 # dispatch 내부 로직을 모사하여 _create_task_docs가 호출 안 됨을 확인 _level_to_int = {"normal": 2, "critical": 3, "security": 4} level_str = "normal" docs_level = _level_to_int.get(level_str, 2) called = [] original_fn = dispatch_mod._create_task_docs def _tracking_fn(tid, lv): called.append((tid, lv)) return original_fn(tid, lv) dispatch_mod._create_task_docs = _tracking_fn try: if docs_level >= 3: dispatch_mod._create_task_docs(task_id, docs_level) finally: dispatch_mod._create_task_docs = original_fn # level=2(normal)이므로 호출되지 않아야 함 assert len(called) == 0, "normal 레벨에서 _create_task_docs가 호출되면 안 됨" # 디렉토리가 생성되지 않아야 함 expected_dir = tmp_path / "memory" / "plans" / "tasks" / task_id assert not expected_dir.exists(), f"normal 레벨에서 디렉토리가 생성되면 안 됨: {expected_dir}" # ------------------------------------------------------------------ # 테스트 4: 동일 task_id로 재실행 → 기존 파일 덮어쓰기 안 됨 # ------------------------------------------------------------------ def test_no_overwrite_existing_docs(self, dispatch_mod, tmp_path): """이미 존재하는 파일은 덮어쓰지 않음""" task_id = "task-1874" # 1차 호출: 파일 생성 dispatch_mod._create_task_docs(task_id, level=3) expected_dir = tmp_path / "memory" / "plans" / "tasks" / task_id plan_file = expected_dir / "plan.md" assert plan_file.exists() # 파일 내용을 임의로 수정 original_content = plan_file.read_text(encoding="utf-8") modified_content = original_content + "\n# 추가된 내용" plan_file.write_text(modified_content, encoding="utf-8") # 2차 호출: 덮어쓰기 시도 result2 = dispatch_mod._create_task_docs(task_id, level=3) # 반환값은 여전히 Path (디렉토리 존재) assert result2 is not None # 파일 내용이 수정된 채로 유지되어야 함 (덮어쓰기 금지) after_content = plan_file.read_text(encoding="utf-8") assert after_content == modified_content, "기존 파일이 덮어써짐: 덮어쓰기 금지 정책 위반" # ------------------------------------------------------------------ # 테스트 5: 잘못된 task_id → None 반환 # ------------------------------------------------------------------ def test_invalid_task_id_rejected(self, dispatch_mod, tmp_path): """path traversal 등 잘못된 task_id는 None 반환""" invalid_ids = [ "../etc/passwd", # path traversal "task-1872/../../../etc", # path traversal with task prefix "task_1872", # 허용되지 않는 언더스코어 시작 "Task-1872", # 대문자 "task-abc", # 숫자 아닌 문자 "task-", # 숫자 없음 "", # 빈 문자열 "task-1872 malicious", # 공백 포함 ] for bad_id in invalid_ids: result = dispatch_mod._create_task_docs(bad_id, level=3) assert result is None, f"잘못된 task_id '{bad_id}'에 대해 None이 반환되어야 함, 실제: {result}" # 디렉토리가 생성되지 않아야 함 tasks_root = tmp_path / "memory" / "plans" / "tasks" if tasks_root.exists(): created_dirs = list(tasks_root.iterdir()) assert len(created_dirs) == 0, f"잘못된 task_id로 디렉토리가 생성됨: {created_dirs}"