"""tests/handoff/test_validate_handoff.py validate_handoff.py CLI 검증 테스트 (task-2458 Phase 2-B) 테스트 케이스: - test_validate_missing_handoff — handoff JSON 미존재 → exit 1 - test_validate_schema_invalid — schema invalid (필수 필드 누락) → exit 1 - test_validate_task_id_mismatch — JSON task_id != --task → exit 1 - test_validate_branch_mismatch — JSON current_branch != --branch → exit 1 - test_validate_head_sha_mismatch — JSON head_sha != git HEAD → exit 1 - test_validate_changed_paths_violation — changed_paths가 allowed_paths 외 → exit 1 - test_validate_pending_work_too_long — pending_work 4001자 → exit 1 - test_validate_pass — 모든 조건 충족 → exit 0 """ from __future__ import annotations import json import os import subprocess from pathlib import Path import pytest # --------------------------------------------------------------------------- # 경로 상수 # --------------------------------------------------------------------------- WORKTREE = Path("/home/jay/workspace/.worktrees/task-2458-dev4") VALIDATE_SCRIPT = WORKTREE / "scripts" / "validate_handoff.py" SCHEMA_PATH = WORKTREE / "memory" / "specs" / "handoff-schema.json" # 테스트용 task-id (스키마 패턴 ^task-\d+$ 만족) TEST_TASK_ID = "task-9010" TEST_BRANCH = f"task/{TEST_TASK_ID}-dev4" # --------------------------------------------------------------------------- # 유효한 handoff 데이터 템플릿 # --------------------------------------------------------------------------- VALID_HANDOFF: dict = { "task_id": TEST_TASK_ID, "schema_version": "1.0", "previous_bot": "dev4", "current_branch": TEST_BRANCH, "base_sha": "abc1234", "head_sha": "def5678", "changed_paths": ["tests/handoff/test_validate_handoff.py"], "allowed_paths": ["tests/handoff/", "scripts/"], "forbidden_paths": [], "test_results": {}, "handoff_reason": "takeover_request", "created_at": "2026-05-05T12:00:00Z", "pending_work": "잔여 작업 없음", } # --------------------------------------------------------------------------- # 헬퍼 # --------------------------------------------------------------------------- def _run_validate( args: list[str], workspace_root: Path, extra_env: dict | None = None, ) -> subprocess.CompletedProcess: """validate_handoff.py 실행 헬퍼.""" env = dict(os.environ) env["WORKSPACE_ROOT"] = str(workspace_root) if extra_env: env.update(extra_env) return subprocess.run( ["python3", str(VALIDATE_SCRIPT)] + args, cwd=str(WORKTREE), capture_output=True, text=True, timeout=30, env=env, ) def _write_handoff(workspace: Path, task_id: str, data: dict) -> Path: """tmp workspace 내에 handoff JSON 파일 기록 후 경로 반환.""" hdir = workspace / "memory" / "handoffs" hdir.mkdir(parents=True, exist_ok=True) # 스키마 파일도 복사 (validate_handoff 가 필요) spec_dir = workspace / "memory" / "specs" spec_dir.mkdir(parents=True, exist_ok=True) schema_dst = spec_dir / "handoff-schema.json" if not schema_dst.exists(): schema_dst.write_bytes(SCHEMA_PATH.read_bytes()) hpath = hdir / f"{task_id}.json" hpath.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return hpath # --------------------------------------------------------------------------- # test_validate_missing_handoff # --------------------------------------------------------------------------- def test_validate_missing_handoff(tmp_path: Path): """handoff JSON이 존재하지 않을 때 exit 1.""" # spec 디렉토리만 준비 (handoff JSON 없음) spec_dir = tmp_path / "memory" / "specs" spec_dir.mkdir(parents=True, exist_ok=True) (spec_dir / "handoff-schema.json").write_bytes(SCHEMA_PATH.read_bytes()) result = _run_validate(["--task", TEST_TASK_ID], workspace_root=tmp_path) assert result.returncode == 1, ( f"handoff 없는데 exit 0 (비정상)\nstdout: {result.stdout}\nstderr: {result.stderr}" ) assert "없음" in result.stderr or "not found" in result.stderr.lower() or "FAIL" in result.stderr, ( f"에러 메시지 없음\nstderr: {result.stderr}" ) # --------------------------------------------------------------------------- # test_validate_schema_invalid # --------------------------------------------------------------------------- def test_validate_schema_invalid(tmp_path: Path): """필수 필드가 누락된 handoff JSON → schema validation 실패 → exit 1.""" invalid_data = { "task_id": TEST_TASK_ID, # schema_version, previous_bot 등 필수 필드 누락 "handoff_reason": "interrupt", "created_at": "2026-05-05T12:00:00Z", } _write_handoff(tmp_path, TEST_TASK_ID, invalid_data) result = _run_validate(["--task", TEST_TASK_ID], workspace_root=tmp_path) assert result.returncode == 1, ( f"schema invalid인데 exit 0\nstdout: {result.stdout}\nstderr: {result.stderr}" ) combined = result.stderr + result.stdout assert "FAIL" in combined or "schema" in combined.lower() or "Schema" in combined, ( f"스키마 에러 메시지 없음\nstderr: {result.stderr}" ) # --------------------------------------------------------------------------- # test_validate_task_id_mismatch # --------------------------------------------------------------------------- def test_validate_task_id_mismatch(tmp_path: Path): """JSON task_id != --task 인수 → exit 1.""" data = dict(VALID_HANDOFF) data["task_id"] = "task-9999" # --task task-9010과 불일치 # current_branch도 task_id에 맞게 수정 (schema 패턴 통과용) data["current_branch"] = "task/task-9999-dev4" _write_handoff(tmp_path, TEST_TASK_ID, data) # 파일은 TEST_TASK_ID로 저장 result = _run_validate(["--task", TEST_TASK_ID], workspace_root=tmp_path) assert result.returncode == 1, ( f"task_id 불일치인데 exit 0\nstdout: {result.stdout}\nstderr: {result.stderr}" ) assert "task_id" in result.stderr or "불일치" in result.stderr or "FAIL" in result.stderr, ( f"task_id 불일치 에러 메시지 없음\nstderr: {result.stderr}" ) # --------------------------------------------------------------------------- # test_validate_branch_mismatch # --------------------------------------------------------------------------- def test_validate_branch_mismatch(tmp_path: Path): """JSON current_branch != --branch 인수 → exit 1.""" data = dict(VALID_HANDOFF) data["current_branch"] = "task/task-9010-dev4" # 올바른 값 _write_handoff(tmp_path, TEST_TASK_ID, data) # --branch 에 다른 값 전달 result = _run_validate( ["--task", TEST_TASK_ID, "--branch", "task/task-9010-dev9"], workspace_root=tmp_path, ) assert result.returncode == 1, ( f"branch 불일치인데 exit 0\nstdout: {result.stdout}\nstderr: {result.stderr}" ) assert "branch" in result.stderr.lower() or "불일치" in result.stderr or "FAIL" in result.stderr, ( f"branch 불일치 에러 메시지 없음\nstderr: {result.stderr}" ) # --------------------------------------------------------------------------- # test_validate_head_sha_mismatch # --------------------------------------------------------------------------- def test_validate_head_sha_mismatch(tmp_path: Path): """JSON head_sha가 실제 git rev-parse 결과와 불일치 → exit 1. 임시 git repo를 tmp_path 내에 만들어 HEAD SHA를 제어한다. """ # 임시 git repo 초기화 repo = tmp_path / "repo" repo.mkdir() subprocess.run(["git", "init", "--initial-branch=main"], cwd=str(repo), check=True, capture_output=True) subprocess.run(["git", "config", "user.email", "test@test.com"], cwd=str(repo), check=True, capture_output=True) subprocess.run(["git", "config", "user.name", "Test"], cwd=str(repo), check=True, capture_output=True) (repo / "init.txt").write_text("init") subprocess.run(["git", "add", "."], cwd=str(repo), check=True, capture_output=True) subprocess.run(["git", "commit", "-m", "init"], cwd=str(repo), check=True, capture_output=True) # 브랜치 생성 + 커밋 subprocess.run( ["git", "checkout", "-b", TEST_BRANCH], cwd=str(repo), check=True, capture_output=True ) (repo / "change.txt").write_text("change") subprocess.run(["git", "add", "."], cwd=str(repo), check=True, capture_output=True) subprocess.run(["git", "commit", "-m", "change"], cwd=str(repo), check=True, capture_output=True) # 실제 HEAD SHA 조회 real_sha = subprocess.run( ["git", "rev-parse", "HEAD"], cwd=str(repo), capture_output=True, text=True ).stdout.strip() # handoff에는 의도적으로 다른 SHA 기록 data = dict(VALID_HANDOFF) data["head_sha"] = "0000000" # 잘못된 SHA data["current_branch"] = TEST_BRANCH # spec/handoff 디렉토리를 repo 내에 구성 spec_dir = repo / "memory" / "specs" spec_dir.mkdir(parents=True, exist_ok=True) (spec_dir / "handoff-schema.json").write_bytes(SCHEMA_PATH.read_bytes()) hdir = repo / "memory" / "handoffs" hdir.mkdir(parents=True, exist_ok=True) hpath = hdir / f"{TEST_TASK_ID}.json" hpath.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") result = _run_validate( ["--task", TEST_TASK_ID, "--check-head-sha"], workspace_root=repo, ) assert result.returncode == 1, ( f"head_sha 불일치인데 exit 0\nstdout: {result.stdout}\nstderr: {result.stderr}" f"\nreal_sha={real_sha}, recorded=0000000" ) assert "head_sha" in result.stderr or "sha" in result.stderr.lower() or "FAIL" in result.stderr, ( f"SHA 불일치 에러 메시지 없음\nstderr: {result.stderr}" ) # --------------------------------------------------------------------------- # test_validate_changed_paths_violation # --------------------------------------------------------------------------- def test_validate_changed_paths_violation(tmp_path: Path): """changed_paths에 allowed_paths 범위 밖 파일 포함 → exit 1.""" data = dict(VALID_HANDOFF) data["allowed_paths"] = ["tests/handoff/"] data["changed_paths"] = [ "tests/handoff/test_foo.py", "scripts/dangerous_script.py", # allowed 범위 밖 ] _write_handoff(tmp_path, TEST_TASK_ID, data) result = _run_validate(["--task", TEST_TASK_ID], workspace_root=tmp_path) assert result.returncode == 1, ( f"changed_paths 위반인데 exit 0\nstdout: {result.stdout}\nstderr: {result.stderr}" ) assert "allowed" in result.stderr.lower() or "초과" in result.stderr or "FAIL" in result.stderr, ( f"경로 위반 에러 메시지 없음\nstderr: {result.stderr}" ) # --------------------------------------------------------------------------- # test_validate_pending_work_too_long # --------------------------------------------------------------------------- def test_validate_pending_work_too_long(tmp_path: Path): """pending_work가 4001자이면 validate에서 FAIL → exit 1. 스키마(maxLength=4000) 또는 validate_handoff 자체 검사 중 하나가 잡아야 한다. 스키마 검사는 JSON 파일 내용 기반이므로, validate_handoff.py 에서 검사하는 경우 파일에 4001자가 직접 기록된 JSON을 제공한다. """ data = dict(VALID_HANDOFF) # 스키마 maxLength=4000 이므로 직접 파일에 써서 bypass 후 # validate_handoff 의 자체 길이 검사(또는 스키마 검사)가 잡는지 확인 data["pending_work"] = "x" * 4001 # 스키마 검증은 maxLength 로 잡힘 → exit 1 기대 # 직접 JSON 파일 작성 (스키마 검증 우회 없이 그냥 씀) spec_dir = tmp_path / "memory" / "specs" spec_dir.mkdir(parents=True, exist_ok=True) (spec_dir / "handoff-schema.json").write_bytes(SCHEMA_PATH.read_bytes()) hdir = tmp_path / "memory" / "handoffs" hdir.mkdir(parents=True, exist_ok=True) hpath = hdir / f"{TEST_TASK_ID}.json" hpath.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") result = _run_validate(["--task", TEST_TASK_ID], workspace_root=tmp_path) assert result.returncode == 1, ( f"pending_work 4001자인데 exit 0\nstdout: {result.stdout}\nstderr: {result.stderr}" ) # --------------------------------------------------------------------------- # test_validate_pass # --------------------------------------------------------------------------- def test_validate_pass(tmp_path: Path): """유효한 handoff JSON → exit 0, stdout에 JSON 출력.""" data = dict(VALID_HANDOFF) _write_handoff(tmp_path, TEST_TASK_ID, data) result = _run_validate(["--task", TEST_TASK_ID], workspace_root=tmp_path) assert result.returncode == 0, ( f"유효한 handoff인데 exit 1\nstdout: {result.stdout}\nstderr: {result.stderr}" ) # stdout이 유효한 JSON인지 확인 try: out_data = json.loads(result.stdout) except json.JSONDecodeError as e: pytest.fail(f"stdout이 유효한 JSON이 아님: {e}\nstdout: {result.stdout}") assert out_data.get("task_id") == TEST_TASK_ID, ( f"출력 JSON의 task_id 불일치: {out_data.get('task_id')}" ) # --------------------------------------------------------------------------- # test_validate_capability_snapshot_forgery # --------------------------------------------------------------------------- def test_validate_capability_snapshot_forgery(tmp_path: Path): """handoff allowed_paths가 task 파일 allowed_resources를 초과(forge)하면 → exit 1. capability snapshot 교차검증: 봇이 handoff JSON에 allowed_paths를 임의로 넓혀 적어 권한을 우회하는 시도를 차단한다. """ # task 파일 생성: scripts/ 만 허용 task_dir = tmp_path / "memory" / "tasks" task_dir.mkdir(parents=True) task_md = task_dir / f"{TEST_TASK_ID}.md" task_md.write_text( """# task-9010 — test ```yaml allowed_resources: paths: - "scripts/" forbidden_paths: - ".github/**" ``` """, encoding="utf-8", ) # handoff 작성자가 forbidden 경로까지 임의로 넓힌 allowed_paths data = dict(VALID_HANDOFF) data["allowed_paths"] = ["scripts/", ".github/workflows/", "secrets/"] data["changed_paths"] = ["scripts/x.py"] # 자체 chk는 통과 _write_handoff(tmp_path, TEST_TASK_ID, data) result = _run_validate(["--task", TEST_TASK_ID], workspace_root=tmp_path) assert result.returncode == 1, ( f"forgery인데 exit 0\nstdout: {result.stdout}\nstderr: {result.stderr}" ) combined = result.stderr + result.stdout assert ( "forgery" in combined.lower() or "초과" in combined or "task 파일" in combined ), f"forgery 에러 메시지 없음\ncombined: {combined}"