""" tests/test_watchdog_chairman_definition.py task-2405 회귀 테스트 — 회장 정의 FALSE positive 5 case 알람 0건 보장 session-watchdog.sh 4 fix(A/B/C/D) 기반 회장 정의 (2026-05-03): TRUE positive(진짜 죽음)만 알람 FALSE positive 절대 0건 5 case: 1. 진행 중 작업 (long-running) 2. 완료 작업 (.done 발급) 3. 회장 인지 (.escalate / .escalate.acked 발급) 4. 후속 task 위임됨 (.superseded_by 마커 또는 다른 task md 언급) 5. 진행 마커 (codex-gate, qc-done, done.merging, pr-creating, external-running) 시나리오: OK 케이스 (skip 검증, 알람 0건): 1. test_escalate_marker_skips_chairman_def 2. test_escalate_acked_skips_chairman_def 3. test_escalate_and_acked_both_skips 4. test_done_file_skips_and_status_promoted 5. test_done_acked_skips_and_status_promoted 6. test_done_notified_skips_and_status_promoted 7. test_done_clear_skips 8. test_superseded_by_marker_skips 9. test_other_task_md_mentions_original_skips 10. test_progress_marker_codex_gate_skips 11. test_progress_marker_qc_done_skips 12. test_progress_marker_done_merging_skips 13. test_progress_marker_pr_creating_skips 14. test_progress_marker_external_running_skips NG 케이스 (TRUE positive 검증, 알람 1건): 15. test_truly_stalled_no_marker_alerts 종합 검증: 16. test_chairman_5_ng_cases_all_zero_alarm """ from __future__ import annotations import json import os import subprocess import time from pathlib import Path # --------------------------------------------------------------------------- # 공통 헬퍼 (test_watchdog_noise_elimination.py 패턴 재사용) # --------------------------------------------------------------------------- ORIG_SCRIPT = Path("/home/jay/workspace/scripts/session-watchdog.sh") def _build_timers(tasks: dict) -> dict: """task-timers.json 포맷.""" return {"tasks": tasks} def _running_task( task_id: str, team_id: str = "dev1-team", task_file: str = "", start_offset: int = -3600, ) -> dict: """status=running 태스크 항목. start_time은 현재 - start_offset 초.""" start_ts = time.strftime( "%Y-%m-%dT%H:%M:%S.000000", time.localtime(time.time() + start_offset), ) return { "task_id": task_id, "team_id": team_id, "status": "running", "start_time": start_ts, "task_file": task_file, "retry_count": 0, "max_retry": 2, } def setup_workspace(tmp_path: Path, timers: dict) -> Path: """ tmp_path 아래 필수 디렉토리·파일을 생성하고 WORKSPACE가 tmp_path를 가리키도록 패치된 스크립트를 반환. """ for d in [ "memory/events", "memory/heartbeats", "memory/tasks", "logs", "scripts", ]: (tmp_path / d).mkdir(parents=True, exist_ok=True) # task-timers.json (tmp_path / "memory" / "task-timers.json").write_text( json.dumps(timers), encoding="utf-8" ) # .env.keys (더미 토큰) (tmp_path / ".env.keys").write_text("ANU_BOT_TOKEN=dummy\n", encoding="utf-8") # task-timer.py 더미 (스크립트 내부 python3 호출 방어) (tmp_path / "memory" / "task-timer.py").write_text( "#!/usr/bin/env python3\n", encoding="utf-8" ) # 스크립트 복사 + WORKSPACE 치환 orig = ORIG_SCRIPT.read_text(encoding="utf-8") patched = orig.replace( 'WORKSPACE="/home/jay/workspace"', f'WORKSPACE="{tmp_path}"', ) script_path = tmp_path / "scripts" / "session-watchdog.sh" script_path.write_text(patched, encoding="utf-8") script_path.chmod(0o755) return script_path def run_watchdog( script_path: Path, tmp_path: Path, extra_env: dict | None = None, cwd: Path | None = None, ) -> tuple[int, str]: """ WATCHDOG_DRY_RUN=1로 스크립트 실행 후 (returncode, log_contents) 반환. """ env = os.environ.copy() env["WATCHDOG_DRY_RUN"] = "1" if extra_env: env.update(extra_env) result = subprocess.run( ["bash", str(script_path)], capture_output=True, text=True, cwd=str(cwd or tmp_path), env=env, timeout=15, ) log_file = tmp_path / "logs" / "session-watchdog.log" log_content = log_file.read_text(encoding="utf-8") if log_file.exists() else "" return result.returncode, log_content def touch_file(path: Path, age_seconds: int = 0) -> None: """파일 생성 후 mtime을 (현재 - age_seconds)로 설정.""" path.touch() if age_seconds > 0: t = time.time() - age_seconds os.utime(path, (t, t)) def _read_status(timers_file: Path, tid: str) -> str: """task-timers.json에서 특정 task의 status 읽기 (Fix B 박제 검증용).""" return json.loads(timers_file.read_text())["tasks"][tid].get("status", "running") # --------------------------------------------------------------------------- # OK 케이스 — 알람 0건 (skip 검증) # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # 1. .escalate 단독 → skip # --------------------------------------------------------------------------- def test_escalate_marker_skips_chairman_def(tmp_path): """ task-2405 회장 정의 Fix A: .escalate 단독 마커 존재 → skip, 로그에 '알람 억제' 포함 (알람 0건). 회장 승인 대기 상태에서는 절대 알람 없음. """ tid = "task-cd001" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) # heartbeat 노후 (stalled 조건 충족) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) # .escalate 마커만 생성 (.acked 없음) touch_file(tmp_path / "memory" / "events" / f"{tid}.escalate") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert "알람 억제" in log or "회장 승인 대기" in log, \ f"escalate 억제 로그 없음: {log}" assert f"{tid}(team=" not in log, \ f"escalate 단독 상태에서 stalled 알람 발생 (회장 정의 위반): {log}" assert "알람 없음" in log or f"{tid}(team=" not in log, \ f"알람 발생 (FALSE positive): {log}" # --------------------------------------------------------------------------- # 2. .escalate.acked 단독 → skip (★ 회장 정의 핵심) # --------------------------------------------------------------------------- def test_escalate_acked_skips_chairman_def(tmp_path): """ task-2405 회장 정의 Fix A 핵심: .escalate.acked 존재 → 회장이 이미 인지 = 알람 그만 → skip. acked = 회장 인지 = 더 이상 알람 불필요. """ tid = "task-cd002" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) # heartbeat 노후 hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) # .escalate.acked 마커만 생성 (.escalate 없어도 acked 단독으로 skip) touch_file(tmp_path / "memory" / "events" / f"{tid}.escalate.acked") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert f"{tid}(team=" not in log, \ f"escalate.acked 상태에서 stalled 알람 발생 (회장 정의 핵심 위반): {log}" # DRY_RUN 알람 본문에 tid 미포함 확인 if "DRY_RUN" in log: assert tid not in log.split("DRY_RUN")[1], \ f"DRY_RUN 알람 본문에 tid 포함됨 (FALSE positive): {log}" # --------------------------------------------------------------------------- # 3. .escalate + .escalate.acked 둘 다 → skip # --------------------------------------------------------------------------- def test_escalate_and_acked_both_skips(tmp_path): """ task-2405 회장 정의 Fix A: .escalate + .escalate.acked 둘 다 있어도 → skip (알람 0건). """ tid = "task-cd003" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) events_dir = tmp_path / "memory" / "events" touch_file(events_dir / f"{tid}.escalate") touch_file(events_dir / f"{tid}.escalate.acked") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert f"{tid}(team=" not in log, \ f"escalate+acked 모두 있는데 stalled 알람 발생 (회장 정의 위반): {log}" # --------------------------------------------------------------------------- # 4. .done 발급 → skip + status=escalated 박제 (Fix B) # --------------------------------------------------------------------------- def test_done_file_skips_and_status_promoted(tmp_path): """ task-2405 회장 정의 Fix B: .done 파일 존재 → 완료 처리 skip + task-timers.json에 status=escalated 박제. """ tid = "task-cd004" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) timers_file = tmp_path / "memory" / "task-timers.json" hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) # .done 마커 생성 touch_file(tmp_path / "memory" / "events" / f"{tid}.done") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert f"{tid}(team=" not in log, \ f".done 있는데 stalled 알람 발생 (회장 정의 위반): {log}" # Fix B: status=escalated 박제 검증 status_after = _read_status(timers_file, tid) assert status_after == "escalated", \ f".done 후 status 박제 실패 (기대: escalated, 실제: {status_after})" # --------------------------------------------------------------------------- # 5. .done.acked → skip + 박제 (Fix B) # --------------------------------------------------------------------------- def test_done_acked_skips_and_status_promoted(tmp_path): """ task-2405 회장 정의 Fix B: .done.acked 존재 → skip + status=escalated 박제. """ tid = "task-cd005" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) timers_file = tmp_path / "memory" / "task-timers.json" hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) touch_file(tmp_path / "memory" / "events" / f"{tid}.done.acked") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert f"{tid}(team=" not in log, \ f".done.acked 있는데 stalled 알람 발생: {log}" status_after = _read_status(timers_file, tid) assert status_after == "escalated", \ f".done.acked 후 status 박제 실패: {status_after}" # --------------------------------------------------------------------------- # 6. .done.notified → skip + 박제 (★ Fix B 신규 마커) # --------------------------------------------------------------------------- def test_done_notified_skips_and_status_promoted(tmp_path): """ task-2405 회장 정의 Fix B 신규 마커: .done.notified 존재 → skip + status=escalated 박제. notified = 완료 알림 발송됨 = 알람 불필요. """ tid = "task-cd006" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) timers_file = tmp_path / "memory" / "task-timers.json" hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) # Fix B 신규 마커: .done.notified touch_file(tmp_path / "memory" / "events" / f"{tid}.done.notified") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert f"{tid}(team=" not in log, \ f".done.notified 있는데 stalled 알람 발생 (Fix B 신규 마커 미처리): {log}" status_after = _read_status(timers_file, tid) assert status_after == "escalated", \ f".done.notified 후 status 박제 실패: {status_after}" # --------------------------------------------------------------------------- # 7. .done.clear → skip # --------------------------------------------------------------------------- def test_done_clear_skips(tmp_path): """ task-2405 회장 정의 Fix B: .done.clear 존재 → 완료 처리 skip (알람 0건). """ tid = "task-cd007" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) touch_file(tmp_path / "memory" / "events" / f"{tid}.done.clear") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert f"{tid}(team=" not in log, \ f".done.clear 있는데 stalled 알람 발생: {log}" # --------------------------------------------------------------------------- # 8. .superseded_by 마커 → skip + 박제 (Fix C) # --------------------------------------------------------------------------- def test_superseded_by_marker_skips(tmp_path): """ task-2405 회장 정의 Fix C: .superseded_by 마커 존재 → 후속 task에 위임됨 → skip + 박제. """ tid = "task-cd008" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) timers_file = tmp_path / "memory" / "task-timers.json" hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) # .superseded_by 마커 생성 (Fix C 1순위) touch_file(tmp_path / "memory" / "events" / f"{tid}.superseded_by") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert f"{tid}(team=" not in log, \ f".superseded_by 마커 있는데 stalled 알람 발생 (Fix C 위반): {log}" # Fix C: status=escalated 박제 검증 (should_skip_for_superseded 처리 여부에 따라) status_after = _read_status(timers_file, tid) assert status_after == "escalated", \ f".superseded_by 후 status 박제 실패: {status_after}" # --------------------------------------------------------------------------- # 9. 다른 task md에 원본 task_id 언급 → skip + 박제 (Fix C 2순위) # --------------------------------------------------------------------------- def test_other_task_md_mentions_original_skips(tmp_path): """ task-2405 회장 정의 Fix C 2순위: 후속 task md 본문에 원본 task_id 언급 → 후속 위임됨으로 판단 → skip + 박제. """ tid = "task-cd009" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) timers_file = tmp_path / "memory" / "task-timers.json" hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) # 원본 task md 생성 (tmp_path / "memory" / "tasks" / f"{tid}.md").write_text( f"# Original task\n---\nteam: dev2-team\n---\n작업 내용\n", encoding="utf-8", ) # 후속 task md에 원본 tid 언급 (Fix C 2순위) (tmp_path / "memory" / "tasks" / "task-9999.md").write_text( f"# 후속 task\n## 배경\n원본: {tid} 실패 후 재작업\n", encoding="utf-8", ) rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert f"{tid}(team=" not in log, \ f"후속 task md 언급인데 stalled 알람 발생 (Fix C 2순위 위반): {log}" status_after = _read_status(timers_file, tid) assert status_after == "escalated", \ f"후속 task md 언급 후 status 박제 실패: {status_after}" # --------------------------------------------------------------------------- # 10. codex-gate 진행 마커 → skip (Fix D / 회장 정의 5번) # --------------------------------------------------------------------------- def test_progress_marker_codex_gate_skips(tmp_path): """ task-2405 회장 정의 5번 (진행 마커): events/.codex-gate 존재 + heartbeat 노후 → alive (알람 0건). long-running 단계 보호 — Codex G1/G2 검증 중. """ tid = "task-cd010" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) touch_file(tmp_path / "memory" / "events" / f"{tid}.codex-gate") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert "alive (long-running)" in log, \ f"codex-gate 마커 alive 로그 없음: {log}" assert f"{tid}(team=" not in log, \ f"codex-gate 진행 마커인데 stalled 알람 발생: {log}" # --------------------------------------------------------------------------- # 11. qc-done 진행 마커 → skip # --------------------------------------------------------------------------- def test_progress_marker_qc_done_skips(tmp_path): """ task-2405 회장 정의 5번: events/.qc-done 존재 + heartbeat 노후 → alive (알람 0건). QC 진행 중 단계 보호. """ tid = "task-cd011" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) touch_file(tmp_path / "memory" / "events" / f"{tid}.qc-done") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert "alive (long-running)" in log, \ f"qc-done 마커 alive 로그 없음: {log}" assert f"{tid}(team=" not in log, \ f"qc-done 진행 마커인데 stalled 알람 발생: {log}" # --------------------------------------------------------------------------- # 12. done.merging 진행 마커 → skip # --------------------------------------------------------------------------- def test_progress_marker_done_merging_skips(tmp_path): """ task-2405 회장 정의 5번: events/.done.merging 존재 + heartbeat 노후 → alive (알람 0건). G3 머지 진행 중 단계 보호. """ tid = "task-cd012" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) touch_file(tmp_path / "memory" / "events" / f"{tid}.done.merging") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert "alive (long-running)" in log, \ f"done.merging 마커 alive 로그 없음: {log}" assert f"{tid}(team=" not in log, \ f"done.merging 진행 마커인데 stalled 알람 발생: {log}" # --------------------------------------------------------------------------- # 13. pr-creating 진행 마커 → skip # --------------------------------------------------------------------------- def test_progress_marker_pr_creating_skips(tmp_path): """ task-2405 회장 정의 5번: events/.pr-creating 존재 + heartbeat 노후 → alive (알람 0건). PR 생성 중 단계 보호 (신규 마커). """ tid = "task-cd013" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) touch_file(tmp_path / "memory" / "events" / f"{tid}.pr-creating") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert "alive (long-running)" in log, \ f"pr-creating 마커 alive 로그 없음: {log}" assert f"{tid}(team=" not in log, \ f"pr-creating 진행 마커인데 stalled 알람 발생: {log}" # --------------------------------------------------------------------------- # 14. external-running 진행 마커 → skip # --------------------------------------------------------------------------- def test_progress_marker_external_running_skips(tmp_path): """ task-2405 회장 정의 5번: events/.external-running 존재 + heartbeat 노후 → alive (알람 0건). 외부 CLI 호출 중 단계 보호 (신규 마커). """ tid = "task-cd014" timers = _build_timers({tid: _running_task(tid, "dev2-team")}) script_path = setup_workspace(tmp_path, timers) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) touch_file(tmp_path / "memory" / "events" / f"{tid}.external-running") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 assert "alive (long-running)" in log, \ f"external-running 마커 alive 로그 없음: {log}" assert f"{tid}(team=" not in log, \ f"external-running 진행 마커인데 stalled 알람 발생: {log}" # --------------------------------------------------------------------------- # NG 케이스 — TRUE positive 검증 (알람 1건) # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # 15. 진짜 죽음 — 마커 모두 부재 → 알람 1건 (★ TRUE positive) # --------------------------------------------------------------------------- def test_truly_stalled_no_marker_alerts(tmp_path): """ task-2405 회장 정의 TRUE positive: 모든 마커 부재 + heartbeat 노후 + events 노후 + PR/worktree 부재 → 알람 1건. ★ 이 케이스만이 진짜 죽음 = 알람이 발생해야 정상. 조건: - .escalate 없음 - .escalate.acked 없음 - .done 계열 없음 - .superseded_by 없음 - 진행 마커 없음 - heartbeat 2000s 노후 (임계 초과) - events 파일 없음 (activity 없음) - gh/git worktree 없음 (DRY_RUN 환경) """ tid = "task-cd015" timers = _build_timers({tid: _running_task(tid, "dev1-team")}) script_path = setup_workspace(tmp_path, timers) # heartbeat 노후 (2000s > 600s 임계) hb = tmp_path / "memory" / "heartbeats" / f"{tid}.heartbeat" touch_file(hb, age_seconds=2000) # 어떠한 마커도 없음 → 진짜 죽음 rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 # STALLED 판정 로그 있어야 함 assert "STALLED" in log, f"TRUE positive 케이스에서 STALLED 판정 없음: {log}" # DRY_RUN 알람 발생 확인 assert "DRY_RUN" in log, f"TRUE positive 케이스에서 알람 미발생 (watchdog 무력화됨): {log}" # tid가 알람 대상에 포함됨 assert tid in log, f"TRUE positive tid가 알람 본문에 없음: {log}" # --------------------------------------------------------------------------- # 종합 검증 — 5 NG case 모두 0건 # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # 16. 회장 정의 5 FALSE positive case 모두 알람 0건 # --------------------------------------------------------------------------- def test_chairman_5_ng_cases_all_zero_alarm(tmp_path): """ task-2405 회장 정의 종합 검증: 5가지 FALSE positive 케이스를 5개 task로 한번에 실행 → 총 알람 0건. task-cd020: 진행 중 작업 (long-running, codex-gate 마커) task-cd021: 완료 작업 (.done 발급) task-cd022: 회장 인지 (.escalate.acked) task-cd023: 후속 task 위임됨 (.superseded_by) task-cd024: 진행 마커 (external-running) """ tid_long_running = "task-cd020" tid_done = "task-cd021" tid_acked = "task-cd022" tid_superseded = "task-cd023" tid_progress = "task-cd024" timers = _build_timers({ tid_long_running: _running_task(tid_long_running, "dev2-team"), tid_done: _running_task(tid_done, "dev2-team"), tid_acked: _running_task(tid_acked, "dev2-team"), tid_superseded: _running_task(tid_superseded, "dev2-team"), tid_progress: _running_task(tid_progress, "dev2-team"), }) script_path = setup_workspace(tmp_path, timers) events_dir = tmp_path / "memory" / "events" hb_dir = tmp_path / "memory" / "heartbeats" # 모든 task heartbeat 노후 (stalled 조건 충족) for tid in [tid_long_running, tid_done, tid_acked, tid_superseded, tid_progress]: touch_file(hb_dir / f"{tid}.heartbeat", age_seconds=2000) # Case 1: 진행 중 작업 (long-running) — codex-gate 마커 touch_file(events_dir / f"{tid_long_running}.codex-gate") # Case 2: 완료 작업 — .done 발급 touch_file(events_dir / f"{tid_done}.done") # Case 3: 회장 인지 — .escalate.acked touch_file(events_dir / f"{tid_acked}.escalate.acked") # Case 4: 후속 task 위임됨 — .superseded_by 마커 touch_file(events_dir / f"{tid_superseded}.superseded_by") # Case 5: 진행 마커 — external-running touch_file(events_dir / f"{tid_progress}.external-running") rc, log = run_watchdog(script_path, tmp_path) assert rc == 0 # 5개 task 모두 알람 없음 검증 for tid in [tid_long_running, tid_done, tid_acked, tid_superseded, tid_progress]: assert f"{tid}(team=" not in log, \ f"[종합검증] {tid} 알람 발생 (FALSE positive — 회장 정의 위반): {log}" # DRY_RUN이라면 알람 본문에 5개 tid 미포함 if "DRY_RUN" in log: dry_run_section = log.split("DRY_RUN")[1] if "DRY_RUN" in log else "" for tid in [tid_long_running, tid_done, tid_acked, tid_superseded, tid_progress]: assert f"{tid}(team=" not in dry_run_section, \ f"[종합검증] DRY_RUN 알람 본문에 {tid} 포함됨 (FALSE positive): {log}" # 총 알람 0건 또는 "알람 없음" 로그 확인 stalled_count = sum( log.count(f"{tid}(team=") for tid in [tid_long_running, tid_done, tid_acked, tid_superseded, tid_progress] ) assert stalled_count == 0, \ f"[종합검증] 5 FALSE positive case 중 {stalled_count}건 알람 발생: {log}"