"""tests/regression/test_done_escalated_conflict.py — task-2471+1 F4 회귀 테스트 (Test 1). ``check_done_escalated_conflict`` 함수와 ``verify_done_preconditions`` 통합 검증. 토르(개발1팀)의 utils/silent_corruption_guard.py 수정을 영구 차단한다. 함수 시그니처: check_done_escalated_conflict(task_id: str, *, events_dir: Optional[str] = None) -> dict verify_done_preconditions(pr_number, repo, *, base_branch, gh_cmd, cwd, task_id, events_dir) -> dict 각 반환값: {"ok": bool, "reason": str, "detail": dict} fail-closed 패턴: 파일 시스템 오류 시 ok=False. 헤임달(개발2팀 테스터) 작성 — task-2471+1 F4. """ from __future__ import annotations import importlib.util import json import sys from pathlib import Path import pytest # --------------------------------------------------------------------------- # 모듈 로드 # --------------------------------------------------------------------------- WORKSPACE = Path(__file__).resolve().parents[2] def _load_module(mod_name: str, file_rel: str): """절대 경로로 모듈 로드 (sys.path 오염 방지).""" file_path = WORKSPACE / file_rel spec = importlib.util.spec_from_file_location(mod_name, str(file_path)) if spec is None or spec.loader is None: raise ImportError(f"cannot load spec for {file_path}") module = importlib.util.module_from_spec(spec) sys.modules[mod_name] = module spec.loader.exec_module(module) return module scg = _load_module( "silent_corruption_guard_conflict_alias", "utils/silent_corruption_guard.py", ) # --------------------------------------------------------------------------- # 필수 함수 존재 여부 검증 (토르 미완료 시 skip) # --------------------------------------------------------------------------- CONFLICT_FN_MISSING = not hasattr(scg, "check_done_escalated_conflict") VERIFY_EXTENDED_MISSING = False # verify_done_preconditions은 항상 존재, task_id 파라미터만 확인 if not CONFLICT_FN_MISSING: import inspect _verify_sig = inspect.signature(scg.verify_done_preconditions) VERIFY_EXTENDED_MISSING = "task_id" not in _verify_sig.parameters SKIP_CONFLICT = pytest.mark.skipif( CONFLICT_FN_MISSING, reason="check_done_escalated_conflict 미구현 (토르 작업 미완료)" ) SKIP_VERIFY_EXTENDED = pytest.mark.skipif( CONFLICT_FN_MISSING or VERIFY_EXTENDED_MISSING, reason="verify_done_preconditions task_id 파라미터 미구현 (토르 작업 미완료)" ) DEFAULT_EVENTS_DIR = str(WORKSPACE / "memory" / "events") def _make_done(events_dir: Path, task_id: str, content: str = "done") -> Path: """events_dir에 .done 파일 생성.""" p = events_dir / f"{task_id}.done" p.write_text(content, encoding="utf-8") return p def _make_escalated(events_dir: Path, task_id: str, content: str = "") -> Path: """events_dir에 .done.escalated 파일 생성.""" p = events_dir / f"{task_id}.done.escalated" p.write_text(content, encoding="utf-8") return p # --------------------------------------------------------------------------- # Case 1: .done 만 존재 → ok=True # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_done_only_ok(tmp_path: Path) -> None: """.done 만 있고 .done.escalated 없으면 ok=True (정상 완료 상태).""" _make_done(tmp_path, "task-9001") result = scg.check_done_escalated_conflict("task-9001", events_dir=str(tmp_path)) assert isinstance(result, dict), "반환값이 dict 이어야 함" assert result["ok"] is True assert "reason" in result assert "detail" in result # --------------------------------------------------------------------------- # Case 2: .done.escalated 만 존재 → ok=True # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_escalated_only_ok(tmp_path: Path) -> None: """.done.escalated 만 있고 .done 없으면 ok=True (에스컬레이션 완료 상태).""" _make_escalated(tmp_path, "task-9002", content=json.dumps({"trigger": "t", "reason": "r"})) result = scg.check_done_escalated_conflict("task-9002", events_dir=str(tmp_path)) assert result["ok"] is True # --------------------------------------------------------------------------- # Case 3: 둘 다 부재 → ok=True # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_both_absent_ok(tmp_path: Path) -> None: """.done / .done.escalated 모두 없으면 ok=True (미완료 상태, 충돌 없음).""" result = scg.check_done_escalated_conflict("task-9003", events_dir=str(tmp_path)) assert result["ok"] is True # --------------------------------------------------------------------------- # Case 4: 둘 다 존재 (정상 .done + 빈 .done.escalated) → ok=False, reason에 "coexist" # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_done_and_empty_escalated_conflict(tmp_path: Path) -> None: """정상 .done + 빈 .done.escalated 동시 존재 → conflict → ok=False, reason='coexist' 포함.""" _make_done(tmp_path, "task-9004", content="done") _make_escalated(tmp_path, "task-9004", content="") # 빈 파일 (결함 marker) result = scg.check_done_escalated_conflict("task-9004", events_dir=str(tmp_path)) assert result["ok"] is False assert "coexist" in result["reason"].lower(), ( f"reason에 'coexist' 없음: {result['reason']!r}" ) # --------------------------------------------------------------------------- # Case 5: 둘 다 존재 (정상 .done + 정상 .done.escalated) → ok=False # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_done_and_valid_escalated_conflict(tmp_path: Path) -> None: """정상 .done + 정상 payload .done.escalated 동시 존재 → conflict → ok=False.""" _make_done(tmp_path, "task-9005", content="done") valid_payload = json.dumps({ "trigger": "done-watcher.sh:stale_done_30min", "ts": "2026-05-07T00:00:00Z", "source": "scripts/done-watcher.sh", "host": "testhost", "done_path": str(tmp_path / "task-9005.done"), "age_seconds": 1900, "reason": "stale .done >= 1800s", }) _make_escalated(tmp_path, "task-9005", content=valid_payload) result = scg.check_done_escalated_conflict("task-9005", events_dir=str(tmp_path)) assert result["ok"] is False, "정상 .done + 정상 .done.escalated 동시 존재는 conflict 이어야 함" # --------------------------------------------------------------------------- # Case 6: events_dir 명시 (tmp_path) → 정상 동작 # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_events_dir_explicit_tmp_path(tmp_path: Path) -> None: """events_dir=str(tmp_path) 명시 시 해당 경로에서 파일 탐색.""" # tmp_path 외 다른 경로의 파일은 무시되어야 함 _make_done(tmp_path, "task-9006") # events_dir을 다른 임시 디렉토리로 지정 → task-9006.done 없는 경로 other_dir = tmp_path / "other" other_dir.mkdir() result_other = scg.check_done_escalated_conflict( "task-9006", events_dir=str(other_dir) ) assert result_other["ok"] is True # other_dir에 파일 없음 → 충돌 없음 # 이번엔 tmp_path 자체로 → .done 존재, 충돌 없음 (escalated 없음) result_ok = scg.check_done_escalated_conflict( "task-9006", events_dir=str(tmp_path) ) assert result_ok["ok"] is True # --------------------------------------------------------------------------- # Case 7: events_dir 부재 (default) → DEFAULT_EVENTS_DIR 사용 # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_events_dir_default_path() -> None: """events_dir 미지정 시 /home/jay/workspace/memory/events 기본값 사용 확인. 기본 이벤트 디렉토리가 실제로 존재하지 않아도 ok=True (파일 없음 = 충돌 없음). task_id가 실제 .done을 생성하지 않으면 ok=True를 반환해야 한다. """ # 실제 존재하지 않는 task_id 사용 → 기본 events_dir에 파일 없음 result = scg.check_done_escalated_conflict("task-nonexistent-heimdall-test-9007") # 충돌이 없어야 한다 (존재 안 하는 파일 = ok) assert result["ok"] is True # --------------------------------------------------------------------------- # Case 8: detail dict 구조 검증 (task_id, done_path, escalated_path, done_exists, escalated_exists) # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_detail_dict_contains_required_keys(tmp_path: Path) -> None: """detail dict에 task_id, done_path, escalated_path, done_exists, escalated_exists 포함 검증.""" _make_done(tmp_path, "task-9008") result = scg.check_done_escalated_conflict("task-9008", events_dir=str(tmp_path)) detail = result.get("detail", {}) required_keys = {"task_id", "done_path", "escalated_path", "done_exists", "escalated_exists"} missing = required_keys - set(detail.keys()) assert not missing, f"detail에 누락된 키: {missing!r}\n실제 detail: {detail!r}" assert detail["task_id"] == "task-9008" assert detail["done_exists"] is True assert detail["escalated_exists"] is False # --------------------------------------------------------------------------- # Case 8b: detail dict 충돌 케이스 검증 # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_detail_dict_conflict_case(tmp_path: Path) -> None: """충돌 케이스에서 detail에 done_exists=True, escalated_exists=True 검증.""" _make_done(tmp_path, "task-9008b") _make_escalated(tmp_path, "task-9008b", content="") result = scg.check_done_escalated_conflict("task-9008b", events_dir=str(tmp_path)) assert result["ok"] is False detail = result.get("detail", {}) assert detail.get("done_exists") is True assert detail.get("escalated_exists") is True assert detail.get("task_id") == "task-9008b" # --------------------------------------------------------------------------- # Case 9: verify_done_preconditions + task_id 명시 → check_done_escalated_conflict 통합 검증 # --------------------------------------------------------------------------- @SKIP_VERIFY_EXTENDED def test_verify_done_preconditions_task_id_conflict_detected( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """verify_done_preconditions(task_id=...) 호출 시 conflict 검사 통합 검증. .done + .done.escalated 충돌 상황에서 verify_done_preconditions가 check_done_escalated_conflict를 내부적으로 호출하여 ok=False를 반환해야 한다. PR 검사는 monkeypatch로 우회하지 않고, 대신 conflict가 먼저 감지되어 PR 검사 전에 실패하는 경우를 검증한다. Note: conflict 검사가 PR 검사보다 먼저 실행되는 구현을 가정. 만약 PR 검사가 먼저라면 PR 검사 실패 후 conflict도 포함된 detail 검증. """ # Conflict 조건 설정 _make_done(tmp_path, "task-9009") _make_escalated(tmp_path, "task-9009", content="") # check_done_escalated_conflict를 직접 단위 테스트하여 conflict 감지 확인 conflict_result = scg.check_done_escalated_conflict( "task-9009", events_dir=str(tmp_path) ) assert conflict_result["ok"] is False, ( "check_done_escalated_conflict가 conflict를 감지해야 함" ) assert "coexist" in conflict_result["reason"].lower(), ( f"reason에 'coexist' 없음: {conflict_result['reason']!r}" ) # verify_done_preconditions에 task_id + events_dir 전달 # gh_cmd는 mock하여 PR 검사가 실패해도 ok=False를 기대 # (conflict 선검사 또는 PR 검사 실패 모두 ok=False여야 함) def _fake_run(*_args, **_kwargs): # noqa: ANN002, ANN003 del _args, _kwargs return (-1, "", "gh: not available in test") monkeypatch.setattr(scg, "_run", _fake_run) if hasattr(scg, "time"): monkeypatch.setattr(scg.time, "sleep", lambda _: None) result = scg.verify_done_preconditions( pr_number=999, repo="owner/repo", base_branch="main", task_id="task-9009", events_dir=str(tmp_path), ) assert result["ok"] is False, ( f"conflict 상황에서 verify_done_preconditions가 ok=True를 반환함: {result!r}" ) # --------------------------------------------------------------------------- # Case 10: check_done_escalated_conflict 단독 — task_id only (경계값) # --------------------------------------------------------------------------- @SKIP_CONFLICT def test_task_id_with_special_chars_handled(tmp_path: Path) -> None: """task_id에 '+' 등 특수문자 포함 시 정상 처리 (파일시스템 안전).""" task_id = "task-2471+1" _make_done(tmp_path, task_id) result = scg.check_done_escalated_conflict(task_id, events_dir=str(tmp_path)) assert isinstance(result, dict) assert "ok" in result assert result["ok"] is True # escalated 없음 → 충돌 없음 # --------------------------------------------------------------------------- # Smoke: 토르 작업 미완료 시 함수 부재 여부 명시 보고 # --------------------------------------------------------------------------- def test_smoke_check_done_escalated_conflict_exists() -> None: """check_done_escalated_conflict 함수가 silent_corruption_guard에 존재해야 함. 토르 작업 미완료 시 이 테스트가 FAIL하여 CI에서 즉시 감지 가능. """ assert hasattr(scg, "check_done_escalated_conflict"), ( "check_done_escalated_conflict 함수 미존재 — 토르의 utils/silent_corruption_guard.py 수정 필요" )