""" test_task_2352_cancel.py 카마소츠 (개발7팀 테스터) 작성 — task-2352 취소 신호 강제 처리 자동 검증 테스트 케이스: 1. test_cancel_task_invalid_id_returns_error 2. test_cancel_task_creates_cancelled_marker 3. test_cancel_task_idempotent_stop_marker 4. test_finish_task_blocks_on_cancelled_marker 5. test_finish_task_detects_top_stop_marker 6. test_qc_verify_handle_gate_skips_on_cancelled """ import json import subprocess import sys from pathlib import Path import pytest # workspace sys.path 보정 _WORKSPACE = Path(__file__).parent.parent sys.path.insert(0, str(_WORKSPACE)) # qc_verify 경로 보정 _SHARED = _WORKSPACE / "teams" / "shared" sys.path.insert(0, str(_SHARED)) import dispatch _TASKS_DIR = _WORKSPACE / "memory" / "tasks" _EVENTS_DIR = _WORKSPACE / "memory" / "events" # =========================================================================== # Test 1: 잘못된 task_id → status=="error" 반환 # =========================================================================== def test_cancel_task_invalid_id_returns_error(): """task_id 형식이 틀린 경우 status=="error"를 반환해야 한다.""" result = dispatch.cancel_task("not-valid") assert result["status"] == "error", f"예상 'error', 실제: {result['status']}" # =========================================================================== # Test 2: cancel_task 호출 시 .cancelled 마커 및 task md CANCELLED 마커 생성 # =========================================================================== def test_cancel_task_creates_cancelled_marker(): """cancel_task 호출 시 .cancelled 이벤트 파일과 task md CANCELLED 마커가 생성된다.""" task_id = "task-99950" task_file = _TASKS_DIR / f"{task_id}.md" cancelled_file = _EVENTS_DIR / f"{task_id}.cancelled" # 사전 정리 (혹시 이전 실행 잔재) task_file.unlink(missing_ok=True) cancelled_file.unlink(missing_ok=True) # 더미 task 파일 생성 _TASKS_DIR.mkdir(parents=True, exist_ok=True) task_file.write_text("# Test\n", encoding="utf-8") try: result = dispatch.cancel_task(task_id) # .cancelled 파일 존재 확인 assert cancelled_file.exists(), f".cancelled 파일이 없음: {cancelled_file}" # task md 첫 줄에 CANCELLED 마커 포함 확인 content = task_file.read_text(encoding="utf-8") assert "CANCELLED" in content.splitlines()[0], ( f"task md 첫 줄에 CANCELLED 없음. 첫 줄: {content.splitlines()[0]!r}" ) # 결과 status=="ok" assert result["status"] == "ok", f"예상 'ok', 실제: {result['status']}" # actions에 stop_marker_added, cancelled_event 포함 assert "stop_marker_added" in result["actions"], "actions에 stop_marker_added 없음" assert "cancelled_event" in result["actions"], "actions에 cancelled_event 없음" finally: task_file.unlink(missing_ok=True) cancelled_file.unlink(missing_ok=True) # =========================================================================== # Test 3: 이미 STOP 마커가 있는 task 파일 → 마커 중복 삽입 안 함 # =========================================================================== def test_cancel_task_idempotent_stop_marker(): """task md에 이미 STOP 마커가 있는 경우, cancel_task 재호출 시 마커가 중복 삽입되지 않는다.""" task_id = "task-99951" task_file = _TASKS_DIR / f"{task_id}.md" cancelled_file = _EVENTS_DIR / f"{task_id}.cancelled" # 사전 정리 task_file.unlink(missing_ok=True) cancelled_file.unlink(missing_ok=True) # 더미 task 파일에 이미 STOP 마커 삽입 stop_marker = "★★★ 작업 취소됨 (CANCELLED) ★★★" _TASKS_DIR.mkdir(parents=True, exist_ok=True) task_file.write_text(f"{stop_marker}\n\n# Test\n", encoding="utf-8") try: dispatch.cancel_task(task_id) # head -3 내에 CANCELLED가 1번만 등장하는지 확인 content = task_file.read_text(encoding="utf-8") head3 = content.splitlines()[:3] cancelled_count = sum(1 for line in head3 if "CANCELLED" in line) assert cancelled_count <= 1, ( f"CANCELLED 마커가 head-3 내에 {cancelled_count}번 등장 (중복 삽입 발생)" ) finally: task_file.unlink(missing_ok=True) cancelled_file.unlink(missing_ok=True) # =========================================================================== # Test 4: finish-task.sh — .cancelled 마커 있으면 .done 생성 차단 # =========================================================================== def test_finish_task_blocks_on_cancelled_marker(): """finish-task.sh 실행 시 .cancelled 마커가 있으면 .done 파일이 생성되지 않는다.""" task_id = "task-99952" task_file = _TASKS_DIR / f"{task_id}.md" cancelled_file = _EVENTS_DIR / f"{task_id}.cancelled" done_file = _EVENTS_DIR / f"{task_id}.done" # 사전 정리 task_file.unlink(missing_ok=True) cancelled_file.unlink(missing_ok=True) done_file.unlink(missing_ok=True) # 더미 task 파일 생성 _TASKS_DIR.mkdir(parents=True, exist_ok=True) _EVENTS_DIR.mkdir(parents=True, exist_ok=True) task_file.write_text("# Test task-99952\n", encoding="utf-8") # .cancelled 미리 생성 cancelled_data = {"task_id": task_id, "cancelled_at": "2026-05-02T00:00:00", "reason": "test"} cancelled_file.write_text(json.dumps(cancelled_data), encoding="utf-8") try: result = subprocess.run( ["bash", str(_WORKSPACE / "scripts" / "finish-task.sh"), task_id], capture_output=True, text=True, ) # returncode == 0 assert result.returncode == 0, ( f"finish-task.sh가 비정상 종료 (returncode={result.returncode})\n" f"stderr: {result.stderr}" ) # stdout/stderr에 [CANCELLED] 또는 '.cancelled 마커 발견' 포함 combined = result.stdout + result.stderr assert "[CANCELLED]" in combined or ".cancelled 마커 발견" in combined, ( f"CANCELLED 메시지가 출력에 없음.\nstdout: {result.stdout}\nstderr: {result.stderr}" ) # .done 파일이 생성되지 않음 assert not done_file.exists(), f".done 파일이 생성되었음 (차단 실패): {done_file}" finally: task_file.unlink(missing_ok=True) cancelled_file.unlink(missing_ok=True) done_file.unlink(missing_ok=True) # =========================================================================== # Test 5: finish-task.sh — task md 상단 STOP 마커 감지 시 .cancelled 자동 생성, .done 미생성 # =========================================================================== def test_finish_task_detects_top_stop_marker(): """task md 첫 줄에 STOP 마커가 있으면 finish-task.sh가 .cancelled를 생성하고 .done은 만들지 않는다.""" task_id = "task-99953" task_file = _TASKS_DIR / f"{task_id}.md" cancelled_file = _EVENTS_DIR / f"{task_id}.cancelled" done_file = _EVENTS_DIR / f"{task_id}.done" # 사전 정리 task_file.unlink(missing_ok=True) cancelled_file.unlink(missing_ok=True) done_file.unlink(missing_ok=True) # 더미 task 파일에 STOP 마커를 첫 줄로 삽입 _TASKS_DIR.mkdir(parents=True, exist_ok=True) _EVENTS_DIR.mkdir(parents=True, exist_ok=True) task_file.write_text( "★★★ 작업 취소됨 (CANCELLED) ★★★\n\n# Test task-99953\n", encoding="utf-8", ) try: result = subprocess.run( ["bash", str(_WORKSPACE / "scripts" / "finish-task.sh"), task_id], capture_output=True, text=True, ) # returncode == 0 assert result.returncode == 0, ( f"finish-task.sh가 비정상 종료 (returncode={result.returncode})\n" f"stderr: {result.stderr}" ) # .cancelled 자동 생성 확인 assert cancelled_file.exists(), ( f"STOP 마커 감지 후 .cancelled 파일이 생성되지 않음: {cancelled_file}" ) # .done 파일이 생성되지 않음 assert not done_file.exists(), f".done 파일이 생성되었음 (차단 실패): {done_file}" finally: task_file.unlink(missing_ok=True) cancelled_file.unlink(missing_ok=True) done_file.unlink(missing_ok=True) # =========================================================================== # Test 6: qc_verify._handle_gate — .cancelled 있으면 .qc-result 미생성, overall=="CANCELLED" # =========================================================================== def test_qc_verify_handle_gate_skips_on_cancelled(monkeypatch): """_handle_gate 호출 시 .cancelled 마커가 있으면 .qc-result가 생성되지 않고 overall이 CANCELLED로 변경된다.""" import qc_verify task_id = "task-99954" cancelled_file = _EVENTS_DIR / f"{task_id}.cancelled" qc_result_file = _EVENTS_DIR / f"{task_id}.qc-result" # 사전 정리 cancelled_file.unlink(missing_ok=True) qc_result_file.unlink(missing_ok=True) # WORKSPACE_ROOT 환경변수를 실제 workspace로 고정 monkeypatch.setenv("WORKSPACE_ROOT", str(_WORKSPACE)) # .cancelled 미리 생성 _EVENTS_DIR.mkdir(parents=True, exist_ok=True) cancelled_data = {"task_id": task_id, "cancelled_at": "2026-05-02T00:00:00", "reason": "test"} cancelled_file.write_text(json.dumps(cancelled_data), encoding="utf-8") dummy_result = {"overall": "PASS", "checks": {}} try: qc_verify._handle_gate(dummy_result, task_id, "dev7") # .qc-result 파일이 생성되지 않음 assert not qc_result_file.exists(), ( f".qc-result 파일이 생성되었음 (.cancelled 가드 실패): {qc_result_file}" ) # result["overall"] == "CANCELLED" assert dummy_result["overall"] == "CANCELLED", ( f"overall이 CANCELLED가 아님: {dummy_result['overall']}" ) finally: cancelled_file.unlink(missing_ok=True) qc_result_file.unlink(missing_ok=True)