"""task-2395 회귀 테스트. 검증 대상: 1. archive_old_done_clear: 90일 이상 파일만 이동 (dummy 10개 검증) 2. archive_old_done_clear: 90일 미만 파일은 원위치 유지 3. check_phase_progress: 6개 중 N개 .done 진행률 정확 4. run_completion_checks: 6/6 도착 시 보고서 파일 존재 확인 분기 5. watch_once: 자동 dispatch.py 호출 시도 차단 (mock 검증) 6. archive: 삭제 함수 부재 확인 (history 보존 의무) """ import os import sys import inspect from datetime import datetime, timedelta, timezone from pathlib import Path SCRIPTS_DIR = Path("/home/jay/workspace/scripts") if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) import ids_phase_monitor as monitor # type: ignore[import-not-found] KST = timezone(timedelta(hours=9)) NOW = datetime(2026, 5, 3, 12, 0, tzinfo=KST) def _make_done_clear(dir_: Path, name: str, days_ago: int, now: datetime = NOW) -> Path: p = dir_ / f"{name}.done.clear" p.write_text("", encoding="utf-8") target = (now - timedelta(days=days_ago)).timestamp() os.utime(p, (target, target)) return p def test_archive_moves_files_older_than_90_days(tmp_path): events_dir = tmp_path / "events" archive_root = events_dir / "archive" events_dir.mkdir() old_files = [_make_done_clear(events_dir, f"task-old-{i}", 91, NOW) for i in range(5)] new_files = [_make_done_clear(events_dir, f"task-new-{i}", 30, NOW) for i in range(5)] result = monitor.archive_old_done_clear(events_dir, archive_root, age_days=90, now=NOW) assert result["moved"] == 5 assert result["skipped"] == 5 assert sum(result["by_month"].values()) == 5 for p in old_files: assert not p.exists(), f"{p} should have been moved" for p in new_files: assert p.exists(), f"{p} should remain" moved_count = 0 for month_dir in archive_root.iterdir(): if month_dir.is_dir(): moved_count += sum(1 for _ in month_dir.glob("*.done.clear")) assert moved_count == 5 def test_archive_preserves_files_under_90_days(tmp_path): events_dir = tmp_path / "events" archive_root = events_dir / "archive" events_dir.mkdir() files = [_make_done_clear(events_dir, f"task-{i}", 80, NOW) for i in range(5)] result = monitor.archive_old_done_clear(events_dir, archive_root, age_days=90, now=NOW) assert result["moved"] == 0 assert result["skipped"] == 5 assert result["by_month"] == {} for p in files: assert p.exists() if archive_root.exists(): assert not any(archive_root.rglob("*.done.clear")) def test_check_phase_progress_partial_and_complete(tmp_path): events_dir = tmp_path / "events" events_dir.mkdir() (events_dir / "task-2389.done").write_text("") (events_dir / "task-2390.dev1.done").write_text("") (events_dir / "task-2391.dev2.done").write_text("") result = monitor.check_phase_progress(events_dir) assert result["n_done"] == 3 assert result["n_total"] == 6 assert result["ratio"] == "3/6" assert result["complete"] is False assert sorted(result["done"]) == ["task-2389", "task-2390", "task-2391"] assert sorted(result["pending"]) == ["task-2392", "task-2393", "task-2394"] (events_dir / "task-2392.done").write_text("") (events_dir / "task-2393.dev3.done").write_text("") (events_dir / "task-2394.done").write_text("") result2 = monitor.check_phase_progress(events_dir) assert result2["complete"] is True assert result2["ratio"] == "6/6" assert result2["n_done"] == 6 def test_check_phase_progress_ignores_done_clear(tmp_path): events_dir = tmp_path / "events" events_dir.mkdir() (events_dir / "task-2389.done.clear").write_text("") result = monitor.check_phase_progress(events_dir) assert result["n_done"] == 0 assert "task-2389" in result["pending"] def test_run_completion_checks_detects_missing_reports(tmp_path): reports_dir = tmp_path / "reports" reports_dir.mkdir() project_root = tmp_path / "proj" project_root.mkdir() for tid in ["task-2389", "task-2390", "task-2391", "task-2392", "task-2393"]: (reports_dir / f"{tid}.md").write_text(f"# {tid}\n") result = monitor.run_completion_checks(reports_dir, project_root) assert len(result["reports_ok"]) == 5 assert result["reports_missing"] == ["task-2394"] assert result["git_status"] == "no-git" assert "pytest" in result["pytest_recommendation"] assert "batch-ids-master.md" in result["batch_report_target"] def test_watch_once_does_not_invoke_dispatch(tmp_path, monkeypatch): events_dir = tmp_path / "events" events_dir.mkdir() reports_dir = tmp_path / "reports" reports_dir.mkdir() project_root = tmp_path / "proj" project_root.mkdir() state_file = tmp_path / "state.json" import subprocess as _sp calls = [] real_run = _sp.run real_popen = _sp.Popen def spy_run(args, *a, **kw): calls.append(("run", args)) if isinstance(args, list) and "dispatch.py" in str(args): return _sp.CompletedProcess(args, 0, "", "") return real_run(args, *a, **kw) def spy_popen(args, *a, **kw): calls.append(("popen", args)) return real_popen(args, *a, **kw) def spy_call(args, *_a, **_kw): calls.append(("call", args)) return 0 monkeypatch.setattr(_sp, "run", spy_run) monkeypatch.setattr(_sp, "Popen", spy_popen) monkeypatch.setattr(_sp, "call", spy_call) monkeypatch.setattr(monitor, "send_telegram_notification", lambda *_a, **_kw: True) monitor.watch_once( events_dir=events_dir, reports_dir=reports_dir, project_root=project_root, state_file=state_file, notify=False, ) for kind, args in calls: s = " ".join(map(str, args)) if isinstance(args, (list, tuple)) else str(args) assert "dispatch.py" not in s, f"dispatch.py invoked via {kind}: {args}" src = inspect.getsource(monitor) assert "import dispatch" not in src assert "from dispatch" not in src def test_module_has_no_delete_functions(): src = inspect.getsource(monitor) forbidden_patterns = [ "os.remove(", "os.unlink(", ".unlink()", "shutil.rmtree(", ] for pattern in forbidden_patterns: assert pattern not in src, f"Forbidden delete pattern found: {pattern}" public_names = [n for n in dir(monitor) if not n.startswith("_")] for name in public_names: assert not name.startswith("delete_"), f"delete_* function found: {name}" assert not name.startswith("remove_"), f"remove_* function found: {name}"