# -*- coding: utf-8 -*- """tests.regression.test_real_merge_executor_no_op_paths — task-2637. 7 NO_OP fixture finalize_result 단언 + 실 subprocess 호출 0. Spec: memory/specs/system_real_merge_executor_wiring_spec_260523.md §5 / §10 """ from __future__ import annotations import json from datetime import datetime from pathlib import Path from typing import Any, List import pytest from utils.real_merge_artifact_schema import ( EXECUTION_RESULT_FILENAME, resolve_artifact_dir, validate_post_smoke_payload, ) from utils.real_merge_hooks import ( NO_OP_FLAG_DISABLED, NO_OP_GATE_FAIL, NO_OP_NO_AUTHORIZATION, NO_OP_RESULTS, NO_OP_STALE_SNAPSHOT, real_merge_execute, ) FIXTURE_ROOT = ( Path(__file__).resolve().parent.parent / "fixtures" / "real_merge_wiring" ) NO_OP_SCENARIOS = [ ("no_op_no_authorization", NO_OP_NO_AUTHORIZATION), ("would_merge_but_disabled", NO_OP_FLAG_DISABLED), ("stale_snapshot", NO_OP_STALE_SNAPSHOT), ("chair_auth_expired", NO_OP_NO_AUTHORIZATION), ("admin_override_required", NO_OP_GATE_FAIL), ("blocking_secret_detected", NO_OP_GATE_FAIL), ] def _load(scenario: str): base = FIXTURE_ROOT / scenario evidence = json.loads((base / "evidence.json").read_text(encoding="utf-8")) expected = json.loads((base / "expected.json").read_text(encoding="utf-8")) return evidence, expected def _kst_clock(text: str): dt = datetime.strptime(text, "%Y-%m-%dT%H:%M:%S%z") return lambda: dt def _runner_must_never_run(*_args: Any, **_kwargs: Any) -> Any: raise AssertionError( "subprocess_runner was invoked during a NO_OP path — " "task-2637 safety invariant violated" ) @pytest.mark.parametrize("scenario,expected_enum", NO_OP_SCENARIOS) def test_no_op_fixture(scenario, expected_enum, tmp_path): evidence, expected = _load(scenario) clock = _kst_clock(evidence["frozen_now_kst"]) if "frozen_now_kst" in evidence else None canonical_root = str(tmp_path) out = real_merge_execute( merge_ready_result=evidence.get("merge_ready_result"), pr_identity=evidence["pr_identity"], gate_snapshot=evidence.get("gate_snapshot"), dryrun_artifact=evidence.get("dryrun_artifact"), callback_envelope=evidence.get("callback_envelope"), chair_authorization=evidence.get("chair_authorization"), activation_flag=bool(evidence["activation_flag"]), subprocess_runner=_runner_must_never_run, canonical_root=canonical_root, changed_files=evidence.get("changed_files"), clock=clock, ) assert out["result_enum"] == expected_enum == expected["result_enum"] assert out["result_enum"] in NO_OP_RESULTS assert out["decision"]["actually_executed"] is False assert out["execution_result"] is None assert out["execution_result_path"] is None assert out["subprocess_invocations"] == [] assert bool(out.get("chair_report_required", False)) == bool( expected.get("chair_report_required", False) ) # Decision reason keyword coverage. reasons_text = " ".join(out["decision"]["reasons"]).lower() for kw in expected.get("decision_reason_keywords", []): assert kw.lower() in reasons_text, ( f"missing keyword {kw!r} in reasons: {out['decision']['reasons']}" ) # Artifact write verification. pr = evidence["pr_identity"]["pr"] head_sha = evidence["pr_identity"]["head_sha"] artifact_dir = Path(resolve_artifact_dir(pr, head_sha, canonical_root)) for fname in expected.get("artifacts_written", []): assert (artifact_dir / fname).is_file(), ( f"expected artifact {fname!r} not written under {artifact_dir}" ) # NO_OP must not write merge_execution_result.json. assert not (artifact_dir / EXECUTION_RESULT_FILENAME).exists() def test_subprocess_runner_zero_calls_across_all_no_op(tmp_path): """Aggregate guarantee — across all 6 NO_OP fixtures the injected runner is never invoked, so the global "실 gh pr merge 0" invariant holds.""" seen_calls: List[Any] = [] def trap_runner(*args: Any, **kwargs: Any) -> Any: seen_calls.append((args, kwargs)) raise AssertionError("runner must not be called in NO_OP path") for scenario, _enum in NO_OP_SCENARIOS: evidence, _ = _load(scenario) clock = _kst_clock(evidence["frozen_now_kst"]) if "frozen_now_kst" in evidence else None out = real_merge_execute( merge_ready_result=evidence.get("merge_ready_result"), pr_identity=evidence["pr_identity"], gate_snapshot=evidence.get("gate_snapshot"), dryrun_artifact=evidence.get("dryrun_artifact"), callback_envelope=evidence.get("callback_envelope"), chair_authorization=evidence.get("chair_authorization"), activation_flag=bool(evidence["activation_flag"]), subprocess_runner=trap_runner, canonical_root=str(tmp_path / scenario), changed_files=evidence.get("changed_files"), clock=clock, ) assert out["subprocess_invocations"] == [] assert seen_calls == [] def test_post_smoke_fail_report_only_schema_only(): """post_smoke_fail_report_only fixture — writer schema-only (writer 본 task 범위 외).""" evidence, expected = _load("post_smoke_fail_report_only") payload = evidence["post_smoke_payload"] errors = validate_post_smoke_payload(payload) assert errors == [], f"unexpected schema errors: {errors}" assert payload["all_ok"] is False == expected["all_ok"] # ANCHOR — 회장 결정 #4: 자동 rollback 금지 · 보고 only. assert payload.get("rollback_recommended") is True assert expected["auto_rollback_invoked"] is False def test_chair_report_required_set_on_admin_override_and_secret(tmp_path): """Two fixtures must surface chair_report_required=True (spec §10).""" for scenario in ("admin_override_required", "blocking_secret_detected"): evidence, _ = _load(scenario) clock = _kst_clock(evidence["frozen_now_kst"]) out = real_merge_execute( merge_ready_result=evidence.get("merge_ready_result"), pr_identity=evidence["pr_identity"], gate_snapshot=evidence.get("gate_snapshot"), dryrun_artifact=evidence.get("dryrun_artifact"), callback_envelope=evidence.get("callback_envelope"), chair_authorization=evidence.get("chair_authorization"), activation_flag=True, subprocess_runner=_runner_must_never_run, canonical_root=str(tmp_path / scenario), changed_files=evidence.get("changed_files"), clock=clock, ) assert out["chair_report_required"] is True assert out["subprocess_invocations"] == []