""" task-2384 Phase β 신호등 통합 — 회귀 테스트 스위트 테스터: 모리건(Morrigan), dev3 QA 6개 테스트: 1. test_resolver_label_to_verdict_mapping — VERDICT_MAP + classify_status 2. test_sweep_stale_completed_batches — _sweep_stale_completed 배치 스윕 3. test_checklist_auto_check_lv3_task — Lv.3 체크리스트 자동 [x] 4. test_checklist_no_op_for_lv0_to_lv2 — Lv.0~2는 체크리스트 no-op 5. test_resolver_failure_falls_back_to_status — resolve_task_verdict 폴백 6. test_concurrent_sweep_lock_safety — _sync_task_timers 동시성 flock 안전 """ import importlib.util import json import subprocess import sys import threading from pathlib import Path import pytest # --------------------------------------------------------------------------- # Import: scripts 경로 설정 # --------------------------------------------------------------------------- sys.path.insert(0, "/home/jay/workspace/scripts") sys.path.insert(0, "/home/jay/workspace") # report_parser 등 workspace 루트 from auto_merge import ( # type: ignore[import-not-found] # noqa: E402 _sweep_stale_completed, _sync_task_timers, _update_plan_checklist, ) import bot_status_resolver as bsr # type: ignore[import-not-found] # noqa: E402 # whisper-compile.py (하이픈 포함 — 직접 import 불가) _wc_spec = importlib.util.spec_from_file_location( "whisper_compile_mod", "/home/jay/workspace/scripts/whisper-compile.py", ) whisper_compile_mod = importlib.util.module_from_spec(_wc_spec) # type: ignore[arg-type] _wc_spec.loader.exec_module(whisper_compile_mod) # type: ignore[union-attr] resolve_task_verdict = whisper_compile_mod.resolve_task_verdict # =========================================================================== # Test 1 — VERDICT_MAP + classify_status # =========================================================================== def test_resolver_label_to_verdict_mapping(): """VERDICT_MAP 5개 라벨-버딕트 매핑 + classify_status 대표 입력값 검증.""" # --- VERDICT_MAP --- expected_map = { "MERGED": "completed", "ALIVE": "in_progress", "IDLE": "in_progress", "STALE": "stalled", "UNKNOWN": "unknown", } for label, verdict in expected_map.items(): assert bsr.VERDICT_MAP[label] == verdict, ( f"VERDICT_MAP[{label!r}] expected {verdict!r}, got {bsr.VERDICT_MAP[label]!r}" ) # --- classify_status 대표 케이스 --- # ALIVE: ps_alive=True, age=2, no PR assert bsr.classify_status( ps_alive=True, last_commit_age_min=2, pr_merged_at=None, pr_number=None ) == "ALIVE" # IDLE: ps_alive=True, age=10, pr_number=42 (하지만 merged_at=None → IDLE) assert bsr.classify_status( ps_alive=True, last_commit_age_min=10, pr_merged_at=None, pr_number=42 ) == "IDLE" # STALE: ps_alive=False, age=60, no PR assert bsr.classify_status( ps_alive=False, last_commit_age_min=60, pr_merged_at=None, pr_number=None ) == "STALE" # MERGED: ps_alive=False, pr_merged_at set → MERGED 우선 assert bsr.classify_status( ps_alive=False, last_commit_age_min=10, pr_merged_at="2026-05-03", pr_number=42 ) == "MERGED" # =========================================================================== # Test 2 — _sweep_stale_completed 배치 스윕 # =========================================================================== def test_sweep_stale_completed_batches(tmp_path): """머지 커밋 메시지에 묶인 task들을 completed로 스윕하는지 검증.""" # task-timers.json 생성 memory_dir = tmp_path / "memory" memory_dir.mkdir(parents=True) timers = { "tasks": { "task-1001": {"status": "running", "team_id": "dev3"}, "task-1002": {"status": "running", "team_id": "dev3"}, "task-1003": {"status": "completed", "team_id": "dev3"}, } } timers_path = memory_dir / "task-timers.json" timers_path.write_text(json.dumps(timers, ensure_ascii=False, indent=2), encoding="utf-8") # fake git repo 초기화 fakeproject = tmp_path / "fakeproject" fakeproject.mkdir() def git(*args, **kwargs): return subprocess.run( ["git", *args], cwd=str(fakeproject), capture_output=True, text=True, **kwargs, ) git("init") git("config", "user.email", "test@test.com") git("config", "user.name", "Tester") dummy = fakeproject / "dummy.txt" dummy.write_text("hello") git("add", "dummy.txt") git("commit", "-m", "chore: bundle task-1001 task-1002 task-1003 [merge]") sha_result = git("rev-parse", "HEAD") sha = sha_result.stdout.strip() assert sha, "git rev-parse HEAD 실패" # 함수 호출 swept = _sweep_stale_completed( workspace=tmp_path, merge_commit_sha=sha, primary_task_id="task-1001", project_path=str(fakeproject), ) # primary는 제외, 이미 completed인 task-1003도 제외 → task-1002만 반환 assert swept == ["task-1002"], f"swept 목록 오류: {swept}" # task-timers.json 재검증 updated = json.loads(timers_path.read_text(encoding="utf-8")) tasks = updated["tasks"] assert tasks["task-1002"]["status"] == "completed", "task-1002가 completed로 갱신되지 않음" assert tasks["task-1001"]["status"] == "running", "primary task-1001은 변경되면 안 됨" # =========================================================================== # Test 3 — _update_plan_checklist: Lv.3 task 자동 [x] # =========================================================================== def _setup_checklist_workspace(tmp_path: Path) -> None: """Test 3/4 공통 픽스처 구성. exist_ok=True로 반복 호출 안전.""" # memory/tasks/task-2000.md tasks_dir = tmp_path / "memory" / "tasks" tasks_dir.mkdir(parents=True, exist_ok=True) (tasks_dir / "task-2000.md").write_text( "---\nproject: dev-system\nlevel: 3\n---\n\n본문 내용", encoding="utf-8", ) # memory/plans/dev-system/checklist.md (매 호출마다 초기 내용으로 리셋) plans_dir = tmp_path / "memory" / "plans" / "dev-system" plans_dir.mkdir(parents=True, exist_ok=True) (plans_dir / "checklist.md").write_text( "# Plan checklist\n" "- [ ] task-2000: implement traffic light fix\n" "- [ ] task-2001: unrelated\n", encoding="utf-8", ) def test_checklist_auto_check_lv3_task(tmp_path): """Lv.3 task 머지 시 checklist.md의 해당 줄을 [x]로 체크.""" _setup_checklist_workspace(tmp_path) result = _update_plan_checklist(tmp_path, "task-2000", level=3) assert result is True, "_update_plan_checklist(level=3) should return True" checklist = (tmp_path / "memory" / "plans" / "dev-system" / "checklist.md").read_text(encoding="utf-8") lines = checklist.splitlines() task_2000_line = next((l for l in lines if "task-2000" in l), None) task_2001_line = next((l for l in lines if "task-2001" in l), None) assert task_2000_line is not None, "task-2000 줄이 체크리스트에 없음" assert "- [x]" in task_2000_line, f"task-2000 줄이 [x]로 변경되지 않음: {task_2000_line!r}" assert task_2001_line is not None, "task-2001 줄이 체크리스트에 없음" assert "- [ ]" in task_2001_line, f"task-2001 줄이 변경됨 (no-op이어야 함): {task_2001_line!r}" # =========================================================================== # Test 4 — _update_plan_checklist: Lv.0~2 no-op # =========================================================================== def test_checklist_no_op_for_lv0_to_lv2(tmp_path): """Lv.0, 1, 2 task는 체크리스트를 수정하지 않는다.""" for level in (0, 1, 2): # 매 레벨마다 clean workspace 재구성 _setup_checklist_workspace(tmp_path) result = _update_plan_checklist(tmp_path, "task-2000", level=level) assert result is False, f"_update_plan_checklist(level={level}) should return False" checklist = (tmp_path / "memory" / "plans" / "dev-system" / "checklist.md").read_text( encoding="utf-8" ) task_2000_line = next( (l for l in checklist.splitlines() if "task-2000" in l), None ) assert task_2000_line is not None, f"task-2000 줄 없음 (level={level})" assert "- [ ]" in task_2000_line, ( f"level={level}에서 task-2000 줄이 변경됨 (no-op이어야 함): {task_2000_line!r}" ) # =========================================================================== # Test 5 — resolve_task_verdict 폴백 동작 # =========================================================================== def test_resolver_failure_falls_back_to_status(monkeypatch): """_resolve_bot_status가 None이거나 예외를 던질 때 fallback_status에서 verdict 도출.""" # --- 케이스 A: _resolve_bot_status = None --- monkeypatch.setattr(whisper_compile_mod, "_resolve_bot_status", None) assert resolve_task_verdict("task-XYZ", "running", {}) == "in_progress" assert resolve_task_verdict("task-XYZ2", "completed", {}) == "completed" assert resolve_task_verdict("task-XYZ3", "", {}) == "unknown" # --- 케이스 B: _resolve_bot_status가 RuntimeError 발생 --- def _exploding_resolver(*args, **kwargs): del args, kwargs raise RuntimeError("boom") monkeypatch.setattr(whisper_compile_mod, "_resolve_bot_status", _exploding_resolver) # 캐시가 다른 task_id이어야 fallback이 호출됨 assert resolve_task_verdict("task-BOOM", "running", {}) == "in_progress" # =========================================================================== # Test 6 — _sync_task_timers 동시성 flock 안전 # =========================================================================== def test_concurrent_sweep_lock_safety(tmp_path): """두 스레드가 동시에 _sync_task_timers를 호출해도 task-timers.json이 손상되지 않는다.""" memory_dir = tmp_path / "memory" memory_dir.mkdir(parents=True) timers = { "tasks": { "task-3001": {"status": "running", "team_id": "dev3"}, "task-3002": {"status": "running", "team_id": "dev3"}, } } timers_path = memory_dir / "task-timers.json" timers_path.write_text(json.dumps(timers, ensure_ascii=False, indent=2), encoding="utf-8") barrier = threading.Barrier(2) errors: list[Exception] = [] def worker(task_id: str, sha: str): try: barrier.wait() # 두 스레드를 동시에 출발 _sync_task_timers(tmp_path, task_id, sha, "test") except Exception as exc: errors.append(exc) t1 = threading.Thread(target=worker, args=("task-3001", "sha1")) t2 = threading.Thread(target=worker, args=("task-3002", "sha2")) t1.start() t2.start() t1.join(timeout=10) t2.join(timeout=10) assert not errors, f"스레드 예외 발생: {errors}" # task-timers.json 유효 JSON인지 확인 raw = timers_path.read_text(encoding="utf-8") try: data = json.loads(raw) except json.JSONDecodeError as exc: pytest.fail(f"task-timers.json이 유효하지 않은 JSON: {exc}\n내용:\n{raw}") tasks = data.get("tasks", {}) assert tasks["task-3001"]["status"] == "completed", "task-3001이 completed가 아님" assert tasks["task-3002"]["status"] == "completed", "task-3002이 completed가 아님"