"""tests/regression/test_silent_corruption.py — task-2471 회귀 테스트. 토르가 commit ea702b51 에서 추가한 ``utils.silent_corruption_guard`` 의 3 fail-closed check (mergedAt / mergeCommit.oid / origin/main ancestry) 와 ``verify_done_preconditions`` 를 영구 차단한다. 외부 호출 (subprocess, gh, git) 은 monkeypatch 로 fake 처리. 실제 ``gh`` / ``git`` 호출 절대 금지 — 격리된 단위 테스트. 헤임달(개발2팀 테스터) 작성. """ from __future__ import annotations import importlib.util import json import sys from pathlib import Path import pytest # Worktree 루트 동적 해소 (tests/regression/.py → 2단 위) WORKSPACE = Path(__file__).resolve().parents[2] def _load_module(mod_name: str, file_rel: str): """Worktree 의 절대 경로로 모듈을 직접 로드 (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_test_alias", "utils/silent_corruption_guard.py", ) # --------------------------------------------------------------------------- # fake _run helper (subprocess 외부 호출 차단) # --------------------------------------------------------------------------- def _patch_run(monkeypatch: pytest.MonkeyPatch, response_map: dict) -> list: """``silent_corruption_guard._run`` 을 사전 정의된 응답 맵으로 fake. ``response_map`` 은 ``cmd[0:N]`` 의 prefix tuple 또는 임의 매칭 함수를 키로 받아 ``(rc, stdout, stderr)`` 를 반환. 실제 호출 사실을 추적하기 위해 호출 로그 list 를 반환. """ calls: list = [] def _fake_run(cmd, *, cwd=None, timeout=None): # noqa: ANN001 calls.append({"cmd": list(cmd), "cwd": cwd, "timeout": timeout}) # gh / git 분기 if cmd and cmd[0] == "gh": return response_map.get("gh", (0, "{}", "")) if cmd and cmd[0] == "git": sub = cmd[1] if len(cmd) > 1 else "" if sub == "fetch": return response_map.get("git_fetch", (0, "", "")) if sub == "rev-parse": return response_map.get("git_rev_parse", (0, "abc123\n", "")) if sub == "merge-base": return response_map.get("git_merge_base", (0, "", "")) return (0, "", "") monkeypatch.setattr(scg, "_run", _fake_run) # sleep 도 0초로 단축 (race window 모사 시간 절약) monkeypatch.setattr(scg.time, "sleep", lambda _: None) return calls # --------------------------------------------------------------------------- # 1. mergedAt null 시 verify_done_preconditions ok=False # --------------------------------------------------------------------------- def test_mergedAt_null_fails_verify_done(monkeypatch): payload = json.dumps({"mergedAt": None, "mergeCommit": {"oid": "deadbeef"}}) _patch_run(monkeypatch, {"gh": (0, payload, "")}) result = scg.verify_done_preconditions( pr_number=42, repo="owner/repo", base_branch="main" ) assert result["ok"] is False assert result["detail"]["failed_check"] == "merged_at" assert "mergedAt" in result["reason"] or "mergedAt" in result["detail"]["checks"][ "merged_at" ]["reason"] def test_check_pr_merged_at_null_returns_ok_false(monkeypatch): """check_pr_merged_at 단독 호출 시 mergedAt=null → ok=False.""" payload = json.dumps({"mergedAt": None, "mergeCommit": {"oid": "x"}}) _patch_run(monkeypatch, {"gh": (0, payload, "")}) result = scg.check_pr_merged_at(7, "owner/repo") assert result["ok"] is False assert "null" in result["reason"].lower() or "not merged" in result["reason"].lower() # --------------------------------------------------------------------------- # 2. mergeCommit.oid null 시 verify_done_preconditions ok=False # --------------------------------------------------------------------------- def test_merge_commit_oid_null_fails_verify_done(monkeypatch): payload = json.dumps({"mergedAt": "2026-05-07T00:00:00Z", "mergeCommit": None}) _patch_run(monkeypatch, {"gh": (0, payload, "")}) result = scg.verify_done_preconditions( pr_number=99, repo="owner/repo", base_branch="main" ) assert result["ok"] is False assert result["detail"]["failed_check"] == "merge_commit_oid" def test_merge_commit_missing_oid_field(monkeypatch): """mergeCommit 객체는 있지만 oid 키가 없는 경우.""" payload = json.dumps({"mergedAt": "2026-05-07T00:00:00Z", "mergeCommit": {}}) _patch_run(monkeypatch, {"gh": (0, payload, "")}) result = scg.check_pr_merge_commit_oid(11, "owner/repo") assert result["ok"] is False assert "oid" in result["reason"].lower() # --------------------------------------------------------------------------- # 3. ancestry 검증 실패 시 ok=False # --------------------------------------------------------------------------- def test_ancestry_check_fails_when_not_ancestor(monkeypatch): """merge-base --is-ancestor 가 rc!=0 인 경우 fail-closed.""" payload = json.dumps({ "mergedAt": "2026-05-07T00:00:00Z", "mergeCommit": {"oid": "merge_sha_xyz"}, }) _patch_run( monkeypatch, { "gh": (0, payload, ""), "git_fetch": (0, "", ""), "git_rev_parse": (0, "origin_sha_123\n", ""), "git_merge_base": (1, "", "not ancestor"), }, ) result = scg.verify_done_preconditions( pr_number=55, repo="owner/repo", base_branch="main" ) assert result["ok"] is False assert result["detail"]["failed_check"] == "ancestry" def test_ancestry_unstable_origin_sha_fails(monkeypatch): """origin/ SHA 가 2회 fetch 사이 흔들리면 race detected → fail.""" # 매 호출마다 다른 SHA 반환 → unstable 로 감지 sha_iter = iter([ "sha_a\n", "sha_b\n", # 1차 / 2차 (불일치 → 재시도) "sha_c\n", "sha_d\n", # 재시도 1차 / 2차 (또 불일치) ]) def _fake_run(cmd, **_kwargs): # noqa: ANN001 del _kwargs if cmd[0] == "git" and cmd[1] == "rev-parse": return (0, next(sha_iter, "sha_x\n"), "") if cmd[0] == "git" and cmd[1] == "fetch": return (0, "", "") return (0, "", "") monkeypatch.setattr(scg, "_run", _fake_run) monkeypatch.setattr(scg.time, "sleep", lambda _: None) result = scg.check_origin_main_ancestry("merge_sha", base_branch="main") assert result["ok"] is False assert "race" in result["reason"].lower() or "unstable" in result["reason"].lower() # --------------------------------------------------------------------------- # 4. 모든 check PASS 시 ok=True # --------------------------------------------------------------------------- def test_all_checks_pass_yields_ok_true(monkeypatch): merge_sha = "1234567890abcdef1234567890abcdef12345678" payload = json.dumps({ "mergedAt": "2026-05-07T00:00:00Z", "mergeCommit": {"oid": merge_sha}, }) _patch_run( monkeypatch, { "gh": (0, payload, ""), "git_fetch": (0, "", ""), "git_rev_parse": (0, f"{merge_sha}\n", ""), "git_merge_base": (0, "", ""), # is-ancestor PASS }, ) result = scg.verify_done_preconditions( pr_number=200, repo="owner/repo", base_branch="main" ) assert result["ok"] is True assert result["detail"]["merge_commit_sha"] == merge_sha assert result["detail"]["merged_at"] == "2026-05-07T00:00:00Z" assert "checks" in result["detail"] # 3 check 모두 통과 흔적 checks = result["detail"]["checks"] assert checks["merged_at"]["ok"] is True assert checks["merge_commit_oid"]["ok"] is True assert checks["ancestry"]["ok"] is True # --------------------------------------------------------------------------- # 5. gh 명령 실패 시 fail-closed # --------------------------------------------------------------------------- def test_gh_command_failure_fails_closed(monkeypatch): """gh rc != 0 → mergedAt check 실패 → verify_done_preconditions ok=False.""" _patch_run(monkeypatch, {"gh": (1, "", "gh: not authenticated")}) result = scg.verify_done_preconditions( pr_number=1, repo="owner/repo", base_branch="main" ) assert result["ok"] is False assert "gh" in result["reason"] or "merged_at" in result["reason"] def test_gh_subprocess_exception_fails_closed(monkeypatch): """subprocess 예외 (FileNotFoundError 등) 시 fail-closed (ok=False).""" def _raise_run(_cmd, **_kwargs): # noqa: ANN001 del _cmd, _kwargs return (-1, "", "FileNotFoundError: gh not found") monkeypatch.setattr(scg, "_run", _raise_run) monkeypatch.setattr(scg.time, "sleep", lambda _: None) result = scg.check_pr_merged_at(1, "owner/repo") assert result["ok"] is False def test_gh_returns_invalid_json_fails_closed(monkeypatch): """gh stdout 가 JSON 이 아니면 fail-closed.""" _patch_run(monkeypatch, {"gh": (0, "not-json garbage", "")}) result = scg.check_pr_merged_at(2, "owner/repo") assert result["ok"] is False assert "JSON" in result["reason"] or "decode" in result["reason"].lower() # --------------------------------------------------------------------------- # 6. dependency injection 검증: gh_cmd 파라미터 사용 시 # --------------------------------------------------------------------------- def test_gh_cmd_parameter_is_used(monkeypatch): """gh_cmd=["custom-gh"] 전달 시 실제 명령 prefix 가 그것으로 시작해야 함.""" captured: list = [] def _fake_run(cmd, **_kwargs): # noqa: ANN001 del _kwargs captured.append(list(cmd)) return (0, json.dumps({"mergedAt": "x", "mergeCommit": {"oid": "y"}}), "") monkeypatch.setattr(scg, "_run", _fake_run) scg.check_pr_merged_at(5, "owner/repo", gh_cmd=["custom-gh", "--token", "T"]) assert captured, "fake _run 이 호출되지 않았음" cmd0 = captured[0] assert cmd0[:3] == ["custom-gh", "--token", "T"], f"cmd prefix 미일치: {cmd0[:5]}" assert "pr" in cmd0 and "view" in cmd0 # --------------------------------------------------------------------------- # 7. empty merge_commit_sha early-fail 검증 # --------------------------------------------------------------------------- def test_empty_merge_commit_sha_short_circuits(): """check_origin_main_ancestry 가 빈 SHA 받으면 즉시 fail (subprocess 호출 X).""" result = scg.check_origin_main_ancestry("", base_branch="main") assert result["ok"] is False assert "empty" in result["reason"].lower() or "merge_commit_sha" in result["reason"]