""" tests/dev7/test_scope_guard.py dev7 scope-guard pytest 시나리오 5종 (task-2364) 시나리오: 1. 정상 케이스: snapshot 작성 → 정상 diff → exit 0 2. forbidden 위반: codegraph cron 케이스 → exit 1 + scope-violation.json 3. legacy 호환: snapshot 없음 + .allow-no-scope.log 마커 → exit 0 4. snapshot 없음, 마커도 없음: exit 1 5. 시스템 자동 파일 무시: diff에 memory/heartbeats/foo.json만 → exit 0 함수 단위 테스트: - _parse_allowed_resources: yaml 파싱, inline list, 못 찾는 경우 - _save_capability_snapshot: sha256, auto forbidden 보강 """ import json import os import subprocess import sys from pathlib import Path import pytest # workspace root를 sys.path에 추가 WORKSPACE_ROOT = "/home/jay/workspace" WORKTREE_ROOT = "/home/jay/workspace/.worktrees/task-2364-dev7" SCOPE_GUARD_SCRIPT = os.path.join(WORKTREE_ROOT, "scripts", "task-scope-guard.sh") sys.path.insert(0, WORKTREE_ROOT) # --------------------------------------------------------------------------- # 헬퍼 # --------------------------------------------------------------------------- def write_snapshot(caps_dir: Path, task_id: str, snapshot: dict) -> Path: """테스트용 snapshot 파일 작성.""" caps_dir.mkdir(parents=True, exist_ok=True) snap_file = caps_dir / f"{task_id}.json" snap_file.write_text(json.dumps(snapshot), encoding="utf-8") return snap_file def run_scope_guard(task_id: str, diff_content: str, env_workspace: str) -> subprocess.CompletedProcess: """task-scope-guard.sh를 subprocess로 호출.""" import tempfile with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: f.write(diff_content) diff_file = f.name try: env = os.environ.copy() env["WORKSPACE"] = env_workspace result = subprocess.run( ["bash", SCOPE_GUARD_SCRIPT, task_id, diff_file], capture_output=True, text=True, env=env, ) return result finally: os.unlink(diff_file) # --------------------------------------------------------------------------- # 시나리오 1: 정상 케이스 # --------------------------------------------------------------------------- def test_scenario1_pass(tmp_path): """정상 케이스: snapshot 있고 diff 파일이 scope 내 → exit 0.""" task_id = "test-sg-scenario1" caps_dir = tmp_path / "memory" / "capabilities" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) snapshot = { "task_id": task_id, "captured_at": "2026-05-02T17:00:00", "source_sha256": "deadbeef", "allowed_resources": { "paths": ["scripts/finish-task.sh", "memory/plans/bot-capability-system/**"], "forbidden_paths": ["memory/events/*.cron-*"], "commands": [], "merge_policy": "manual", "ttl_hours": 48, }, } write_snapshot(caps_dir, task_id, snapshot) diff_content = "scripts/finish-task.sh\nmemory/plans/bot-capability-system/plan.md\n" result = run_scope_guard(task_id, diff_content, str(tmp_path)) try: assert result.returncode == 0, f"exit 0 기대, got {result.returncode}\nstdout={result.stdout}\nstderr={result.stderr}" assert "PASS" in result.stdout, f"PASS 메시지 기대\nstdout={result.stdout}" finally: # cleanup snap_file = caps_dir / f"{task_id}.json" if snap_file.exists(): snap_file.unlink() # --------------------------------------------------------------------------- # 시나리오 2: forbidden 위반 (codegraph cron 재현) # --------------------------------------------------------------------------- def test_scenario2_forbidden_violation(tmp_path): """forbidden 위반: scope 밖 파일 수정 → exit 1 + scope-violation.json.""" task_id = "test-sg-scenario2" caps_dir = tmp_path / "memory" / "capabilities" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) # task-2360 사고 재현: paths는 scripts/finish-task.sh만, cron 파일 건드림 snapshot = { "task_id": task_id, "captured_at": "2026-05-02T17:00:00", "source_sha256": "deadbeef", "allowed_resources": { "paths": ["scripts/finish-task.sh"], "forbidden_paths": ["memory/events/*.cron-*"], "commands": [], "merge_policy": "manual", "ttl_hours": 48, }, } write_snapshot(caps_dir, task_id, snapshot) # cron-CC712188.json: paths에도 없고 scope 밖 diff_content = "memory/events/cron-CC712188.json\n" result = run_scope_guard(task_id, diff_content, str(tmp_path)) violation_file = events_dir / f"{task_id}.scope-violation.json" try: assert result.returncode == 1, f"exit 1 기대, got {result.returncode}\nstdout={result.stdout}\nstderr={result.stderr}" assert violation_file.exists(), f"scope-violation.json 없음: {violation_file}" with open(violation_file) as f: vdata = json.load(f) assert vdata["task_id"] == task_id assert len(vdata["violations"]) > 0 assert any("cron-CC712188.json" in v["path"] for v in vdata["violations"]) finally: if violation_file.exists(): violation_file.unlink() snap_file = caps_dir / f"{task_id}.json" if snap_file.exists(): snap_file.unlink() # --------------------------------------------------------------------------- # 시나리오 3: legacy 호환 (.allow-no-scope.log 마커) # --------------------------------------------------------------------------- def test_scenario3_legacy_allow_no_scope(tmp_path): """legacy 호환: snapshot 없음 + .allow-no-scope.log → exit 0 + 경고.""" task_id = "test-sg-scenario3" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) # capabilities 디렉토리는 만들되 snapshot 없음 (tmp_path / "memory" / "capabilities").mkdir(parents=True, exist_ok=True) # .allow-no-scope.log 마커 생성 marker_file = events_dir / f"{task_id}.allow-no-scope.log" marker_file.write_text(json.dumps({"task_id": task_id, "reason": "test"})) diff_content = "scripts/finish-task.sh\n" result = run_scope_guard(task_id, diff_content, str(tmp_path)) try: assert result.returncode == 0, f"exit 0 기대, got {result.returncode}\nstdout={result.stdout}\nstderr={result.stderr}" # 경고 메시지 확인 assert "legacy" in result.stderr.lower() or "allow-no-scope" in result.stderr.lower(), \ f"legacy 경고 메시지 없음\nstderr={result.stderr}" finally: if marker_file.exists(): marker_file.unlink() # --------------------------------------------------------------------------- # 시나리오 4: snapshot 없음, 마커도 없음 → exit 1 # --------------------------------------------------------------------------- def test_scenario4_no_snapshot_no_marker(tmp_path): """snapshot 없음 + 마커도 없음 → exit 1 + 에러 메시지.""" task_id = "test-sg-scenario4" # capabilities 디렉토리는 만들되 snapshot 없음 (tmp_path / "memory" / "capabilities").mkdir(parents=True, exist_ok=True) (tmp_path / "memory" / "events").mkdir(parents=True, exist_ok=True) diff_content = "scripts/finish-task.sh\n" result = run_scope_guard(task_id, diff_content, str(tmp_path)) assert result.returncode == 1, f"exit 1 기대, got {result.returncode}\nstdout={result.stdout}\nstderr={result.stderr}" assert "snapshot" in result.stderr.lower() or "dispatch" in result.stderr.lower(), \ f"에러 메시지 없음\nstderr={result.stderr}" # --------------------------------------------------------------------------- # 시나리오 5: 시스템 자동 파일 무시 → exit 0 # --------------------------------------------------------------------------- def test_scenario5_system_files_ignored(tmp_path): """시스템 자동 파일(memory/heartbeats/...)만 diff → exit 0.""" task_id = "test-sg-scenario5" caps_dir = tmp_path / "memory" / "capabilities" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) snapshot = { "task_id": task_id, "captured_at": "2026-05-02T17:00:00", "source_sha256": "deadbeef", "allowed_resources": { "paths": ["scripts/finish-task.sh"], "forbidden_paths": [], "commands": [], "merge_policy": "manual", "ttl_hours": 48, }, } write_snapshot(caps_dir, task_id, snapshot) # 시스템 자동 파일만 포함 diff_content = ( "memory/heartbeats/foo.json\n" "memory/daily/2026-05-02.md\n" "bot-activity.json\n" "token-ledger.json\n" ".heartbeat\n" ) result = run_scope_guard(task_id, diff_content, str(tmp_path)) snap_file = caps_dir / f"{task_id}.json" try: assert result.returncode == 0, f"exit 0 기대, got {result.returncode}\nstdout={result.stdout}\nstderr={result.stderr}" assert "PASS" in result.stdout, f"PASS 메시지 기대\nstdout={result.stdout}" finally: if snap_file.exists(): snap_file.unlink() # --------------------------------------------------------------------------- # 함수 단위 테스트: _parse_allowed_resources # --------------------------------------------------------------------------- def test_parse_allowed_resources_yaml_block(): """fenced yaml 블록에서 allowed_resources 파싱.""" from dispatch import _parse_allowed_resources text = """ # task ## allowed_resources ```yaml allowed_resources: paths: - "scripts/finish-task.sh" - "memory/plans/bot-capability-system/**" forbidden_paths: - "memory/events/*.cron-*" commands: - "pytest" merge_policy: "manual" ttl_hours: 48 ``` """ r = _parse_allowed_resources(text) assert r is not None, "파싱 결과가 None이면 안됨" assert "scripts/finish-task.sh" in r["paths"] assert "memory/plans/bot-capability-system/**" in r["paths"] assert "memory/events/*.cron-*" in r["forbidden_paths"] assert r["merge_policy"] == "manual" assert r["ttl_hours"] == 48 def test_parse_allowed_resources_inline_list(): """인라인 리스트 형식 파싱.""" from dispatch import _parse_allowed_resources text = """ ```yaml allowed_resources: paths: [foo.py, bar.py] forbidden_paths: [secrets/**] merge_policy: manual ttl_hours: 24 ``` """ r = _parse_allowed_resources(text) assert r is not None assert "foo.py" in r["paths"] assert "bar.py" in r["paths"] assert "secrets/**" in r["forbidden_paths"] def test_parse_allowed_resources_not_found(): """allowed_resources 없는 경우 None 반환.""" from dispatch import _parse_allowed_resources text = "# 단순 task 설명\n어떤 작업을 합니다.\n" r = _parse_allowed_resources(text) assert r is None, f"None 기대, got {r}" def test_parse_allowed_resources_yaml_without_key(): """yaml 블록이 있어도 allowed_resources 키 없으면 None.""" from dispatch import _parse_allowed_resources text = """ ```yaml other_key: foo: bar ``` """ r = _parse_allowed_resources(text) assert r is None, f"None 기대, got {r}" # --------------------------------------------------------------------------- # 함수 단위 테스트: _save_capability_snapshot # --------------------------------------------------------------------------- def test_save_capability_snapshot_basic(): """기본 snapshot 저장 + sha256 확인.""" from dispatch import _save_capability_snapshot task_id = "task-test-snap-basic" ar = { "paths": ["scripts/finish-task.sh"], "forbidden_paths": [], "commands": [], "merge_policy": "manual", "ttl_hours": 24, } source_text = "hello world snapshot test" snap_path = _save_capability_snapshot(task_id, ar, source_text) try: assert snap_path.exists(), f"snapshot 파일 없음: {snap_path}" with open(snap_path) as f: data = json.load(f) assert data["task_id"] == task_id assert data["source_sha256"], "sha256 비어있음" # sha256 검증 import hashlib expected = hashlib.sha256(source_text.encode()).hexdigest() assert data["source_sha256"] == expected, f"sha256 불일치: {data['source_sha256']} != {expected}" finally: if snap_path.exists(): snap_path.unlink() def test_save_capability_snapshot_auto_forbidden_capabilities(): """memory/capabilities/** 자동 보강 확인.""" from dispatch import _save_capability_snapshot task_id = "task-test-snap-forbidden" ar = { "paths": ["scripts/finish-task.sh"], "forbidden_paths": ["secrets/**"], # 기존 forbidden "commands": [], "merge_policy": "manual", "ttl_hours": 24, } snap_path = _save_capability_snapshot(task_id, ar, "test source") try: with open(snap_path) as f: data = json.load(f) forbidden = data["allowed_resources"]["forbidden_paths"] assert "memory/capabilities/**" in forbidden, \ f"memory/capabilities/** 자동 보강 없음: {forbidden}" assert "secrets/**" in forbidden, \ f"기존 forbidden_paths 유지 안됨: {forbidden}" finally: if snap_path.exists(): snap_path.unlink() def test_save_capability_snapshot_no_duplicate_forbidden(): """이미 memory/capabilities/**가 있으면 중복 추가 안 함.""" from dispatch import _save_capability_snapshot task_id = "task-test-snap-nodup" ar = { "paths": ["scripts/finish-task.sh"], "forbidden_paths": ["memory/capabilities/**"], # 이미 있음 "commands": [], "merge_policy": "manual", "ttl_hours": 24, } snap_path = _save_capability_snapshot(task_id, ar, "test source") try: with open(snap_path) as f: data = json.load(f) forbidden = data["allowed_resources"]["forbidden_paths"] count = forbidden.count("memory/capabilities/**") assert count == 1, f"중복 추가됨: {forbidden}" finally: if snap_path.exists(): snap_path.unlink()