""" test_whisper_save_guidance.py scripts/whisper-save-guidance.py 단위 테스트 테스트 항목: - 빈 상태 → 기본 guidance 생성 - active 작업 있음 → guidance에 포함 - done 파일 있음 → pending_dispatches에 포함 - idle 팀 있음 → idle_teams에 포함 - 기존 session-guidance.json 존재 시 덮어쓰기 - 파일 손상 시 graceful 처리 """ import importlib.util import json import sys from datetime import datetime, timedelta, timezone from pathlib import Path import pytest # scripts 디렉토리를 import path에 추가 _SCRIPTS_DIR = Path(__file__).parent.parent sys.path.insert(0, str(_SCRIPTS_DIR)) # whisper-save-guidance.py는 하이픈이 있으므로 importlib으로 임포트 _MODULE_PATH = _SCRIPTS_DIR / "whisper-save-guidance.py" spec = importlib.util.spec_from_file_location("whisper_save_guidance", _MODULE_PATH) assert spec is not None whisper_save_guidance = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(whisper_save_guidance) # --------------------------------------------------------------------------- # 헬퍼 함수 # --------------------------------------------------------------------------- def make_bot_activity(bots: dict, base_path: Path) -> Path: """bot-activity.json 픽스처 파일 생성.""" events_dir = base_path / "events" events_dir.mkdir(parents=True, exist_ok=True) path = events_dir / "bot-activity.json" path.write_text(json.dumps({"bots": bots}), encoding="utf-8") return path def make_task_timers(tasks: dict, base_path: Path) -> Path: """task-timers.json 픽스처 파일 생성.""" path = base_path / "task-timers.json" path.write_text(json.dumps({"tasks": tasks}), encoding="utf-8") return path def make_done_file(task_id: str, base_path: Path) -> Path: """*.done 픽스처 파일 생성.""" events_dir = base_path / "events" events_dir.mkdir(parents=True, exist_ok=True) done_path = events_dir / f"{task_id}.done" done_path.write_text("", encoding="utf-8") return done_path def now_iso() -> str: return datetime.now(timezone.utc).isoformat() def hours_ago_iso(hours: float) -> str: dt = datetime.now(timezone.utc) - timedelta(hours=hours) return dt.isoformat() # --------------------------------------------------------------------------- # load_bot_activity() 테스트 # --------------------------------------------------------------------------- class TestLoadBotActivity: """load_bot_activity() 함수 단위 테스트.""" def test_empty_dir_returns_empty_dict(self, tmp_path: Path): """events 디렉토리가 없어도 빈 dict 반환 (graceful fallback).""" result = whisper_save_guidance.load_bot_activity(base_dir=tmp_path) assert result == {} def test_valid_file_parsed(self, tmp_path: Path): """정상 bot-activity.json 파싱.""" bots = {"dev1": {"status": "idle", "since": now_iso()}} make_bot_activity(bots, tmp_path) result = whisper_save_guidance.load_bot_activity(base_dir=tmp_path) assert "dev1" in result assert result["dev1"]["status"] == "idle" def test_corrupted_file_returns_empty_dict(self, tmp_path: Path): """손상된 JSON 파일 → 빈 dict 반환.""" events_dir = tmp_path / "events" events_dir.mkdir(parents=True) (events_dir / "bot-activity.json").write_text("NOT_JSON{{", encoding="utf-8") result = whisper_save_guidance.load_bot_activity(base_dir=tmp_path) assert result == {} def test_missing_bots_key_returns_empty_dict(self, tmp_path: Path): """bots 키가 없는 JSON → 빈 dict 반환.""" events_dir = tmp_path / "events" events_dir.mkdir(parents=True) (events_dir / "bot-activity.json").write_text(json.dumps({"other": 1}), encoding="utf-8") result = whisper_save_guidance.load_bot_activity(base_dir=tmp_path) assert result == {} # --------------------------------------------------------------------------- # load_task_timers() 테스트 # --------------------------------------------------------------------------- class TestLoadTaskTimers: """load_task_timers() 함수 단위 테스트.""" def test_missing_file_returns_empty_dict(self, tmp_path: Path): """task-timers.json 없으면 빈 dict 반환.""" result = whisper_save_guidance.load_task_timers(base_dir=tmp_path) assert result == {} def test_running_task_included(self, tmp_path: Path): """running 상태 작업이 반환 dict에 포함.""" tasks = { "task-565.1": { "task_id": "task-565.1", "team_id": "dev1-team", "description": "테스트 작업", "start_time": now_iso(), "end_time": None, "status": "running", } } make_task_timers(tasks, tmp_path) result = whisper_save_guidance.load_task_timers(base_dir=tmp_path) assert "task-565.1" in result def test_corrupted_file_returns_empty_dict(self, tmp_path: Path): """손상된 JSON → 빈 dict.""" (tmp_path / "task-timers.json").write_text("{bad json", encoding="utf-8") result = whisper_save_guidance.load_task_timers(base_dir=tmp_path) assert result == {} def test_missing_tasks_key_returns_empty_dict(self, tmp_path: Path): """tasks 키 없으면 빈 dict.""" (tmp_path / "task-timers.json").write_text(json.dumps({"other": {}}), encoding="utf-8") result = whisper_save_guidance.load_task_timers(base_dir=tmp_path) assert result == {} # --------------------------------------------------------------------------- # scan_done_files() 테스트 # --------------------------------------------------------------------------- class TestScanDoneFiles: """scan_done_files() 함수 단위 테스트.""" def test_no_events_dir_returns_empty_list(self, tmp_path: Path): """events 디렉토리 없으면 빈 리스트.""" result = whisper_save_guidance.scan_done_files(base_dir=tmp_path) assert result == [] def test_done_file_detected(self, tmp_path: Path): """*.done 파일이 있으면 task_id가 반환 리스트에 포함.""" make_done_file("task-564.1", tmp_path) result = whisper_save_guidance.scan_done_files(base_dir=tmp_path) assert "task-564.1" in result def test_non_done_files_ignored(self, tmp_path: Path): """*.done 이 아닌 파일은 무시.""" events_dir = tmp_path / "events" events_dir.mkdir(parents=True) (events_dir / "bot-activity.json").write_text("{}", encoding="utf-8") (events_dir / "task-100.done.clear").write_text("", encoding="utf-8") # .done.clear는 제외 result = whisper_save_guidance.scan_done_files(base_dir=tmp_path) # task-100 은 .done 파일 아니므로 포함되면 안 됨 assert "task-100" not in result def test_multiple_done_files(self, tmp_path: Path): """여러 .done 파일이 모두 포함.""" make_done_file("task-564.1", tmp_path) make_done_file("task-565.1", tmp_path) result = whisper_save_guidance.scan_done_files(base_dir=tmp_path) assert "task-564.1" in result assert "task-565.1" in result # --------------------------------------------------------------------------- # generate_guidance() 테스트 # --------------------------------------------------------------------------- class TestGenerateGuidance: """generate_guidance() 함수 단위 테스트.""" def test_empty_state_returns_valid_structure(self): """빈 상태에서도 출력 구조가 완전.""" result = whisper_save_guidance.generate_guidance( bots={}, tasks={}, done_task_ids=[], ) assert "last_session" in result assert "guidance" in result assert "pending_dispatches" in result assert "idle_teams" in result assert "active_tasks" in result def test_active_task_appears_in_guidance(self): """running 작업이 guidance 텍스트와 active_tasks에 포함.""" tasks = { "task-565.1": { "task_id": "task-565.1", "team_id": "dev1-team", "description": "테스트", "status": "running", "start_time": now_iso(), "end_time": None, } } result = whisper_save_guidance.generate_guidance( bots={}, tasks=tasks, done_task_ids=[], ) assert "task-565.1" in result["active_tasks"] assert "task-565.1" in result["guidance"] def test_done_task_appears_in_pending_dispatches(self): """done 파일 task_id가 pending_dispatches에 포함.""" result = whisper_save_guidance.generate_guidance( bots={}, tasks={}, done_task_ids=["task-564.1"], ) # pending_dispatches 에 task-564.1 언급 포함 pending_text = " ".join(result["pending_dispatches"]) assert "task-564.1" in pending_text assert "task-564.1" in result["guidance"] def test_idle_team_3h_plus_appears(self): """3시간 이상 idle인 팀이 idle_teams에 포함.""" bots = { "dev2": { "status": "idle", "since": hours_ago_iso(4), # 4시간 전 } } result = whisper_save_guidance.generate_guidance( bots=bots, tasks={}, done_task_ids=[], ) assert any("dev2" in t for t in result["idle_teams"]) def test_idle_team_under_3h_excluded(self): """3시간 미만 idle은 idle_teams에 포함되지 않음.""" bots = { "dev1": { "status": "idle", "since": hours_ago_iso(1), # 1시간 전 } } result = whisper_save_guidance.generate_guidance( bots=bots, tasks={}, done_task_ids=[], ) assert not any("dev1" in t for t in result["idle_teams"]) def test_processing_bot_not_in_idle_teams(self): """processing 상태 봇은 idle_teams에 포함되지 않음.""" bots = { "dev1": { "status": "processing", "since": hours_ago_iso(10), # 오래됐어도 processing } } result = whisper_save_guidance.generate_guidance( bots=bots, tasks={}, done_task_ids=[], ) assert not any("dev1" in t for t in result["idle_teams"]) def test_last_session_is_iso_format(self): """last_session 필드가 ISO8601 형식.""" result = whisper_save_guidance.generate_guidance( bots={}, tasks={}, done_task_ids=[], ) # 파싱 가능한지 확인 dt = datetime.fromisoformat(result["last_session"]) assert dt is not None def test_completed_task_not_in_active_tasks(self): """completed 상태 작업은 active_tasks에 포함되지 않음.""" tasks = { "task-500.1": { "task_id": "task-500.1", "team_id": "dev1-team", "description": "완료된 작업", "status": "completed", "start_time": hours_ago_iso(2), "end_time": hours_ago_iso(1), } } result = whisper_save_guidance.generate_guidance( bots={}, tasks=tasks, done_task_ids=[], ) assert "task-500.1" not in result["active_tasks"] # --------------------------------------------------------------------------- # save_guidance() 테스트 # --------------------------------------------------------------------------- class TestSaveGuidance: """save_guidance() 함수 단위 테스트.""" def test_creates_file(self, tmp_path: Path): """session-guidance.json 파일이 생성됨.""" data = { "last_session": "2026-03-15T02:05:00", "guidance": "테스트 가이던스", "pending_dispatches": [], "idle_teams": [], "active_tasks": [], } whisper_save_guidance.save_guidance(data, base_dir=tmp_path) out_path = tmp_path / "whisper" / "session-guidance.json" assert out_path.exists() def test_written_content_is_valid_json(self, tmp_path: Path): """저장된 파일이 valid JSON.""" data = { "last_session": "2026-03-15T02:05:00", "guidance": "테스트", "pending_dispatches": ["task-1 확인 필요"], "idle_teams": ["dev2-team"], "active_tasks": ["task-565.1"], } whisper_save_guidance.save_guidance(data, base_dir=tmp_path) out_path = tmp_path / "whisper" / "session-guidance.json" loaded = json.loads(out_path.read_text(encoding="utf-8")) assert loaded == data def test_overwrites_existing_file(self, tmp_path: Path): """기존 session-guidance.json이 있으면 덮어씀.""" whisper_dir = tmp_path / "whisper" whisper_dir.mkdir(parents=True) out_path = whisper_dir / "session-guidance.json" out_path.write_text(json.dumps({"old": "data"}), encoding="utf-8") new_data = { "last_session": "2026-03-15T09:00:00", "guidance": "새 가이던스", "pending_dispatches": [], "idle_teams": [], "active_tasks": [], } whisper_save_guidance.save_guidance(new_data, base_dir=tmp_path) loaded = json.loads(out_path.read_text(encoding="utf-8")) assert loaded["guidance"] == "새 가이던스" assert "old" not in loaded def test_creates_whisper_dir_if_missing(self, tmp_path: Path): """whisper 디렉토리가 없으면 자동 생성.""" data = { "last_session": "2026-03-15T02:05:00", "guidance": "", "pending_dispatches": [], "idle_teams": [], "active_tasks": [], } whisper_save_guidance.save_guidance(data, base_dir=tmp_path) assert (tmp_path / "whisper").is_dir() # --------------------------------------------------------------------------- # 통합 테스트: main flow # --------------------------------------------------------------------------- class TestIntegration: """전체 파이프라인 통합 테스트.""" def test_empty_state_no_crash(self, tmp_path: Path): """모든 입력 파일 없어도 정상 완료 및 출력 파일 생성.""" whisper_save_guidance.run(base_dir=tmp_path) out_path = tmp_path / "whisper" / "session-guidance.json" assert out_path.exists() data = json.loads(out_path.read_text(encoding="utf-8")) assert "guidance" in data def test_active_task_in_output(self, tmp_path: Path): """active 작업이 있으면 출력에 포함.""" tasks = { "task-565.1": { "task_id": "task-565.1", "team_id": "dev1-team", "description": "구현 작업", "status": "running", "start_time": now_iso(), "end_time": None, } } make_task_timers(tasks, tmp_path) whisper_save_guidance.run(base_dir=tmp_path) out_path = tmp_path / "whisper" / "session-guidance.json" data = json.loads(out_path.read_text(encoding="utf-8")) assert "task-565.1" in data["active_tasks"] assert "task-565.1" in data["guidance"] def test_done_file_in_output(self, tmp_path: Path): """done 파일이 있으면 pending_dispatches에 포함.""" make_done_file("task-564.1", tmp_path) whisper_save_guidance.run(base_dir=tmp_path) out_path = tmp_path / "whisper" / "session-guidance.json" data = json.loads(out_path.read_text(encoding="utf-8")) pending_text = " ".join(data["pending_dispatches"]) assert "task-564.1" in pending_text def test_idle_team_in_output(self, tmp_path: Path): """3시간 이상 idle 팀이 idle_teams에 포함.""" bots = { "dev3": {"status": "idle", "since": hours_ago_iso(5)}, } make_bot_activity(bots, tmp_path) whisper_save_guidance.run(base_dir=tmp_path) out_path = tmp_path / "whisper" / "session-guidance.json" data = json.loads(out_path.read_text(encoding="utf-8")) assert any("dev3" in t for t in data["idle_teams"]) def test_overwrite_existing_guidance(self, tmp_path: Path): """기존 session-guidance.json 덮어쓰기.""" # 먼저 이전 결과 생성 whisper_save_guidance.run(base_dir=tmp_path) # 새 상태 추가 tasks = { "task-999.1": { "task_id": "task-999.1", "team_id": "dev1-team", "description": "새 작업", "status": "running", "start_time": now_iso(), "end_time": None, } } make_task_timers(tasks, tmp_path) whisper_save_guidance.run(base_dir=tmp_path) out_path = tmp_path / "whisper" / "session-guidance.json" data = json.loads(out_path.read_text(encoding="utf-8")) assert "task-999.1" in data["active_tasks"] def test_corrupted_inputs_graceful(self, tmp_path: Path): """손상된 입력 파일들이 있어도 정상 완료.""" events_dir = tmp_path / "events" events_dir.mkdir(parents=True) (events_dir / "bot-activity.json").write_text("CORRUPT{{{{", encoding="utf-8") (tmp_path / "task-timers.json").write_text("CORRUPT{{{{", encoding="utf-8") # 예외 없이 완료되어야 함 whisper_save_guidance.run(base_dir=tmp_path) out_path = tmp_path / "whisper" / "session-guidance.json" assert out_path.exists()