"""tests/regression/test_empty_escalation_marker.py — task-2471+1 F4 회귀 테스트 (Test 2). ``check_escalation_marker_payload`` 함수와 done-watcher.sh / finish-task.sh 통합 검증. 토르(개발1팀)의 수정이 .done.escalated marker에 올바른 JSON payload를 박제하는지 영구 차단. 함수 시그니처: check_escalation_marker_payload(task_id: str, *, events_dir: Optional[str] = None) -> dict 각 반환값: {"ok": bool, "reason": str, "detail": dict} fail-closed 패턴. A. silent_corruption_guard 측 check_escalation_marker_payload 단위 테스트 (7건) B. done-watcher.sh inline python 통합 검증 (1건+) C. finish-task.sh ESCALATED 분기 검증 (2건) 헤임달(개발2팀 테스터) 작성 — task-2471+1 F4. """ from __future__ import annotations import importlib.util import json import os import subprocess import sys from pathlib import Path import pytest # --------------------------------------------------------------------------- # 모듈 로드 # --------------------------------------------------------------------------- WORKSPACE = Path(__file__).resolve().parents[2] DONE_WATCHER_SCRIPT = WORKSPACE / "scripts" / "done-watcher.sh" FINISH_TASK_SCRIPT = WORKSPACE / "scripts" / "finish-task.sh" 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_payload_alias", "utils/silent_corruption_guard.py", ) # --------------------------------------------------------------------------- # 필수 함수 존재 여부 검증 (토르 미완료 시 skip) # --------------------------------------------------------------------------- PAYLOAD_FN_MISSING = not hasattr(scg, "check_escalation_marker_payload") SKIP_PAYLOAD = pytest.mark.skipif( PAYLOAD_FN_MISSING, reason="check_escalation_marker_payload 미구현 (토르 작업 미완료)" ) REQUIRED_PAYLOAD_KEYS = {"trigger", "ts", "source", "host", "done_path", "age_seconds", "reason"} def _write_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 # =========================================================================== # A. check_escalation_marker_payload 단위 테스트 # =========================================================================== # --------------------------------------------------------------------------- # A-1: 파일 부재 → ok=True # --------------------------------------------------------------------------- @SKIP_PAYLOAD def test_payload_file_absent_ok(tmp_path: Path) -> None: """.done.escalated 파일 없으면 ok=True (아직 에스컬레이션 미발생 = 정상).""" result = scg.check_escalation_marker_payload("task-8001", events_dir=str(tmp_path)) assert isinstance(result, dict) assert result["ok"] is True # --------------------------------------------------------------------------- # A-2: 0 bytes 파일 → ok=False, reason에 "empty" 포함, detail.size=0 # --------------------------------------------------------------------------- @SKIP_PAYLOAD def test_payload_empty_file_fail(tmp_path: Path) -> None: """0 bytes .done.escalated → ok=False, reason에 'empty' 포함, detail.size=0.""" _write_escalated(tmp_path, "task-8002", content="") result = scg.check_escalation_marker_payload("task-8002", events_dir=str(tmp_path)) assert result["ok"] is False assert "empty" in result["reason"].lower(), ( f"reason에 'empty' 없음: {result['reason']!r}" ) detail = result.get("detail", {}) assert detail.get("size") == 0, f"detail.size != 0: {detail.get('size')!r}" # --------------------------------------------------------------------------- # A-3: 비-JSON 텍스트 ("hello") → ok=False, reason에 "JSON" 포함 # --------------------------------------------------------------------------- @SKIP_PAYLOAD def test_payload_non_json_fail(tmp_path: Path) -> None: """비-JSON 텍스트 .done.escalated → ok=False, reason에 'JSON' 포함.""" _write_escalated(tmp_path, "task-8003", content="hello world not json") result = scg.check_escalation_marker_payload("task-8003", events_dir=str(tmp_path)) assert result["ok"] is False assert "json" in result["reason"].lower(), ( f"reason에 'JSON' 없음: {result['reason']!r}" ) # --------------------------------------------------------------------------- # A-4: JSON이지만 list/scalar → ok=False, reason "missing required keys" # --------------------------------------------------------------------------- @SKIP_PAYLOAD def test_payload_json_list_fail(tmp_path: Path) -> None: """JSON array .done.escalated → ok=False (dict 아님).""" _write_escalated(tmp_path, "task-8004a", content="[1, 2, 3]") result = scg.check_escalation_marker_payload("task-8004a", events_dir=str(tmp_path)) assert result["ok"] is False # reason에 "missing required keys" 또는 "not a dict/object" 등 포함 reason_lower = result["reason"].lower() assert ( "missing" in reason_lower or "required" in reason_lower or "dict" in reason_lower or "object" in reason_lower or "invalid" in reason_lower ), f"reason이 부적절: {result['reason']!r}" @SKIP_PAYLOAD def test_payload_json_scalar_fail(tmp_path: Path) -> None: """JSON scalar string .done.escalated → ok=False (dict 아님).""" _write_escalated(tmp_path, "task-8004b", content='"x"') result = scg.check_escalation_marker_payload("task-8004b", events_dir=str(tmp_path)) assert result["ok"] is False # --------------------------------------------------------------------------- # A-5: dict이지만 trigger 누락 → ok=False # --------------------------------------------------------------------------- @SKIP_PAYLOAD def test_payload_dict_missing_trigger_fail(tmp_path: Path) -> None: """trigger 키 누락 .done.escalated → ok=False.""" payload = { # trigger 누락 "ts": "2026-05-07T00:00:00Z", "source": "test", "host": "testhost", "done_path": "/tmp/x.done", "age_seconds": 1900, "reason": "test", } _write_escalated(tmp_path, "task-8005", content=json.dumps(payload)) result = scg.check_escalation_marker_payload("task-8005", events_dir=str(tmp_path)) assert result["ok"] is False # --------------------------------------------------------------------------- # A-6: dict이지만 reason 누락 → ok=False # --------------------------------------------------------------------------- @SKIP_PAYLOAD def test_payload_dict_missing_reason_fail(tmp_path: Path) -> None: """reason 키 누락 .done.escalated → ok=False.""" payload = { "trigger": "done-watcher.sh:stale_done_30min", "ts": "2026-05-07T00:00:00Z", "source": "test", "host": "testhost", "done_path": "/tmp/x.done", "age_seconds": 1900, # reason 누락 } _write_escalated(tmp_path, "task-8006", content=json.dumps(payload)) result = scg.check_escalation_marker_payload("task-8006", events_dir=str(tmp_path)) assert result["ok"] is False # --------------------------------------------------------------------------- # A-7: 정상 payload (trigger + reason 포함) → ok=True, detail.payload_keys 정렬된 키 포함 # --------------------------------------------------------------------------- @SKIP_PAYLOAD def test_payload_valid_full_ok(tmp_path: Path) -> None: """trigger + reason + 모든 필수 키 포함 .done.escalated → ok=True.""" payload = { "trigger": "done-watcher.sh:stale_done_30min", "ts": "2026-05-07T00:00:00Z", "source": "scripts/done-watcher.sh:96-122", "host": "testhost", "done_path": "/home/jay/workspace/memory/events/task-8007.done", "age_seconds": 1900, "reason": "stale .done unprocessed for >= 1800s; cron escalation triggered", } _write_escalated(tmp_path, "task-8007", content=json.dumps(payload)) result = scg.check_escalation_marker_payload("task-8007", events_dir=str(tmp_path)) assert result["ok"] is True detail = result.get("detail", {}) # payload_keys가 정렬된 리스트로 포함되어야 함 payload_keys = detail.get("payload_keys") assert payload_keys is not None, f"detail에 payload_keys 없음: {detail!r}" assert sorted(payload_keys) == payload_keys, ( f"payload_keys가 정렬되지 않음: {payload_keys!r}" ) # trigger, reason이 payload_keys에 포함 assert "trigger" in payload_keys assert "reason" in payload_keys # =========================================================================== # Smoke: 토르 작업 미완료 시 함수 부재 여부 명시 보고 # =========================================================================== def test_smoke_check_escalation_marker_payload_exists() -> None: """check_escalation_marker_payload 함수가 silent_corruption_guard에 존재해야 함.""" assert hasattr(scg, "check_escalation_marker_payload"), ( "check_escalation_marker_payload 함수 미존재 — 토르의 utils/silent_corruption_guard.py 수정 필요" ) # =========================================================================== # B. done-watcher.sh inline python 통합 검증 # =========================================================================== # done-watcher.sh의 stale .done 에스컬레이션 부분 inline python 코드 (복사) _DONE_WATCHER_INLINE_PY = """\ import os, sys, json, datetime, socket path = os.environ.get('ESCALATED_PATH', '') done_path = os.environ.get('ESCALATED_DONE_PATH', '') age_seconds = int(os.environ.get('ESCALATED_AGE', '0')) if not path: sys.exit(1) payload = { 'trigger': 'done-watcher.sh:stale_done_30min', 'ts': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'source': 'scripts/done-watcher.sh:96-122', 'host': socket.gethostname(), 'done_path': done_path, 'age_seconds': age_seconds, 'reason': 'stale .done unprocessed for >= 1800s; cron escalation triggered' } try: fd = os.open(path, os.O_CREAT|os.O_EXCL|os.O_WRONLY, 0o644) with os.fdopen(fd, 'w', encoding='utf-8') as f: json.dump(payload, f, ensure_ascii=False, indent=2) except FileExistsError: sys.exit(1) """ def test_done_watcher_inline_python_creates_valid_payload(tmp_path: Path) -> None: """done-watcher.sh의 inline python이 올바른 JSON payload를 생성하는지 검증. 결과 파일 존재 + 크기 > 0 + JSON 파싱 가능 + 모든 필수 키 포함. """ events_dir = tmp_path / "events" events_dir.mkdir() task_id = "task-test-8100" done_file = events_dir / f"{task_id}.done" escalated_file = events_dir / f"{task_id}.done.escalated" # .done 파일 생성 (실제 mtime 조작은 불필요 — python 코드만 직접 테스트) done_file.write_text("done", encoding="utf-8") env = os.environ.copy() env["ESCALATED_PATH"] = str(escalated_file) env["ESCALATED_DONE_PATH"] = str(done_file) env["ESCALATED_AGE"] = "1900" result = subprocess.run( ["python3", "-c", _DONE_WATCHER_INLINE_PY], env=env, capture_output=True, text=True, timeout=10, ) # 1. 정상 종료 assert result.returncode == 0, ( f"inline python 실패 rc={result.returncode}\nstdout={result.stdout}\nstderr={result.stderr}" ) # 2. 결과 파일 존재 assert escalated_file.exists(), ".done.escalated 파일이 생성되지 않음" # 3. 크기 > 0 file_size = escalated_file.stat().st_size assert file_size > 0, f".done.escalated 파일이 비어 있음 (size={file_size})" # 4. JSON 파싱 가능 try: payload = json.loads(escalated_file.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: pytest.fail(f".done.escalated JSON 파싱 실패: {exc}") # 5. 모든 필수 키 포함 (trigger/ts/source/host/done_path/age_seconds/reason) missing_keys = REQUIRED_PAYLOAD_KEYS - set(payload.keys()) assert not missing_keys, ( f"payload에 누락된 필수 키: {missing_keys!r}\n실제 payload: {payload!r}" ) # 6. 값 검증 assert payload["trigger"] == "done-watcher.sh:stale_done_30min" assert payload["age_seconds"] == 1900 assert payload["done_path"] == str(done_file) assert payload["reason"].startswith("stale") def test_done_watcher_inline_python_idempotent_file_exists(tmp_path: Path) -> None: """이미 .done.escalated가 존재할 때 inline python은 rc=1로 종료 (O_EXCL 보장).""" events_dir = tmp_path / "events" events_dir.mkdir() task_id = "task-test-8101" done_file = events_dir / f"{task_id}.done" escalated_file = events_dir / f"{task_id}.done.escalated" done_file.write_text("done", encoding="utf-8") # 이미 escalated 파일 존재 escalated_file.write_text('{"trigger":"existing"}', encoding="utf-8") env = os.environ.copy() env["ESCALATED_PATH"] = str(escalated_file) env["ESCALATED_DONE_PATH"] = str(done_file) env["ESCALATED_AGE"] = "1900" result = subprocess.run( ["python3", "-c", _DONE_WATCHER_INLINE_PY], env=env, capture_output=True, text=True, timeout=10, ) # FileExistsError → sys.exit(1) → rc=1 assert result.returncode == 1, ( f"O_EXCL 중복 생성 방지 실패 — rc={result.returncode} (기대값=1)" ) # 기존 파일 내용이 보존되어야 함 content = escalated_file.read_text(encoding="utf-8") assert '"existing"' in content, "기존 escalated 내용이 덮어씌워짐" # =========================================================================== # C. finish-task.sh ESCALATED 분기 검증 # =========================================================================== def test_finish_task_sh_syntax_valid() -> None: """finish-task.sh bash syntax 검증 (bash -n).""" assert FINISH_TASK_SCRIPT.exists(), f"finish-task.sh 없음: {FINISH_TASK_SCRIPT}" result = subprocess.run( ["bash", "-n", str(FINISH_TASK_SCRIPT)], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0, ( f"finish-task.sh syntax error:\n{result.stderr}" ) def test_done_watcher_sh_syntax_valid() -> None: """done-watcher.sh bash syntax 검증 (bash -n).""" assert DONE_WATCHER_SCRIPT.exists(), f"done-watcher.sh 없음: {DONE_WATCHER_SCRIPT}" result = subprocess.run( ["bash", "-n", str(DONE_WATCHER_SCRIPT)], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0, ( f"done-watcher.sh syntax error:\n{result.stderr}" ) def test_finish_task_sh_escalated_inline_python_creates_valid_payload(tmp_path: Path) -> None: """finish-task.sh ESCALATED 분기의 inline python이 올바른 JSON payload를 생성하는지 검증. finish-task.sh의 ESCALATED 분기 inline python 코드를 직접 실행하여 .done.escalated payload가 trigger/ts/source/state/task_id/reason 키를 포함하는지 확인. """ # finish-task.sh ESCALATED 분기 inline python (복사) task_id = "task-test-8200" events_dir = tmp_path / "events" events_dir.mkdir() escalated_path = events_dir / f"{task_id}.done.escalated" finish_task_inline_py = f"""\ import json, datetime, sys payload = {{ 'trigger': 'finish-task.sh:taskctl_state_escalated', 'ts': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'source': 'scripts/finish-task.sh:449-454', 'state': 'ESCALATED', 'task_id': '{task_id}', 'reason': 'taskctl status returned ESCALATED; .done blocked, escalation marker emitted' }} with open('{escalated_path}', 'w') as f: json.dump(payload, f, ensure_ascii=False, indent=2) """ result = subprocess.run( ["python3", "-c", finish_task_inline_py], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0, ( f"finish-task.sh ESCALATED inline python 실패: {result.stderr}" ) assert escalated_path.exists(), ".done.escalated 파일 미생성" try: payload = json.loads(escalated_path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: pytest.fail(f"ESCALATED payload JSON 파싱 실패: {exc}") # finish-task.sh ESCALATED 필수 키 검증 required_keys = {"trigger", "ts", "source", "state", "task_id", "reason"} missing_keys = required_keys - set(payload.keys()) assert not missing_keys, ( f"ESCALATED payload 누락 키: {missing_keys!r}\n실제: {payload!r}" ) assert payload["trigger"] == "finish-task.sh:taskctl_state_escalated" assert payload["state"] == "ESCALATED" assert payload["task_id"] == task_id assert "ESCALATED" in payload["reason"]