""" test_task_timer.py memory/task-timer.py 단위 테스트 (아르고스 작성) 테스트 항목: - start_task(): 올바른 JSON 구조 반환 (status="started") - end_task(): 올바른 JSON 구조 반환 (status="completed", duration 포함) - end_task() 존재하지 않는 task_id → status="error" - list_tasks(): 전체 / 상태별 조회 - _format_duration(): 초/분/시간 변환 - add_log_entry(): 일일 로그 파일 생성 확인 격리: TaskTimer(workspace_path=str(tmp_path)) 로 파일시스템 완전 격리 """ import importlib import importlib.util import json import os import sys import time from datetime import datetime from pathlib import Path import pytest # --------------------------------------------------------------------------- # task-timer 모듈 로드 (하이픈 포함 파일명 → importlib 필요) # --------------------------------------------------------------------------- _TIMER_MODULE_PATH = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) / "memory" / "task-timer.py" def _load_task_timer_module(): """하이픈이 포함된 파일명을 importlib.util로 로드""" spec = importlib.util.spec_from_file_location("task_timer", _TIMER_MODULE_PATH) assert spec is not None, f"모듈 스펙을 찾을 수 없음: {_TIMER_MODULE_PATH}" module = importlib.util.module_from_spec(spec) assert spec.loader is not None, "모듈 로더가 없음" spec.loader.exec_module(module) return module _task_timer_mod = _load_task_timer_module() TaskTimer = _task_timer_mod.TaskTimer # --------------------------------------------------------------------------- # fixture: 격리된 TaskTimer 인스턴스 # --------------------------------------------------------------------------- @pytest.fixture() def timer(tmp_path): """tmp_path를 workspace로 사용하는 TaskTimer 인스턴스""" (tmp_path / "memory").mkdir(parents=True, exist_ok=True) return TaskTimer(workspace_path=str(tmp_path)) # --------------------------------------------------------------------------- # 1. start_task() 구조 검증 # --------------------------------------------------------------------------- class TestStartTask: """start_task()가 올바른 JSON 구조를 반환하는지 확인""" def test_returns_status_started(self, timer): result = timer.start_task("task-1.1") assert result["status"] == "started" def test_returns_task_id(self, timer): result = timer.start_task("task-1.1") assert result["task_id"] == "task-1.1" def test_returns_start_time(self, timer): result = timer.start_task("task-1.1") assert "start_time" in result # ISO 형식 파싱 가능한지 확인 datetime.fromisoformat(result["start_time"]) def test_task_written_to_timer_file(self, timer, tmp_path): timer.start_task("task-1.1", team_id="dev1-team", description="테스트 작업") timer_file = tmp_path / "memory" / "task-timers.json" assert timer_file.exists() data = json.loads(timer_file.read_text()) assert "task-1.1" in data["tasks"] def test_task_status_is_running_in_file(self, timer, tmp_path): timer.start_task("task-2.1") timer_file = tmp_path / "memory" / "task-timers.json" data = json.loads(timer_file.read_text()) assert data["tasks"]["task-2.1"]["status"] == "running" def test_start_task_with_team_and_description(self, timer): result = timer.start_task("task-1.1", team_id="dev2-team", description="API 개발") assert result["status"] == "started" def test_start_task_with_project_id(self, timer, tmp_path): timer.start_task("task-1.1", team_id="dev1-team", description="작업", project_id="myproj") timer_file = tmp_path / "memory" / "task-timers.json" data = json.loads(timer_file.read_text()) assert data["tasks"]["task-1.1"]["project_id"] == "myproj" def test_start_task_end_time_is_none_initially(self, timer, tmp_path): timer.start_task("task-1.1") timer_file = tmp_path / "memory" / "task-timers.json" data = json.loads(timer_file.read_text()) assert data["tasks"]["task-1.1"]["end_time"] is None def test_start_task_duration_is_none_initially(self, timer, tmp_path): timer.start_task("task-1.1") timer_file = tmp_path / "memory" / "task-timers.json" data = json.loads(timer_file.read_text()) assert data["tasks"]["task-1.1"]["duration_seconds"] is None class TestStartTaskCompletedGuard: """completed 상태 task 덮어쓰기 방지 테스트 (task-464.1)""" def test_start_completed_task_returns_error(self, timer): """이미 completed된 task_id로 start 시도 → error 반환""" timer.start_task("task-1.1") timer.end_task("task-1.1") result = timer.start_task("task-1.1") assert result["status"] == "error" assert "completed" in result["reason"].lower() or "already completed" in result["reason"].lower() def test_start_completed_task_preserves_original(self, timer, tmp_path): """completed task start 시도 후 원본 데이터 유지""" timer.start_task("task-1.1", team_id="dev1-team", description="원본 작업") timer.end_task("task-1.1") timer.start_task("task-1.1", team_id="dev2-team", description="덮어쓰기 시도") timer_file = tmp_path / "memory" / "task-timers.json" import json data = json.loads(timer_file.read_text()) assert data["tasks"]["task-1.1"]["status"] == "completed" assert data["tasks"]["task-1.1"]["team_id"] == "dev1-team" def test_running_task_still_rejected(self, timer): """기존 running 중복 등록 거부 동작 유지""" timer.start_task("task-1.1") result = timer.start_task("task-1.1") assert result["status"] == "already_running" # --------------------------------------------------------------------------- # 2. end_task() 구조 검증 # --------------------------------------------------------------------------- class TestEndTask: """end_task()가 올바른 JSON 구조를 반환하는지 확인""" def test_returns_status_completed(self, timer): timer.start_task("task-1.1") result = timer.end_task("task-1.1") assert result["status"] == "completed" def test_returns_task_id(self, timer): timer.start_task("task-1.1") result = timer.end_task("task-1.1") assert result["task_id"] == "task-1.1" def test_returns_start_time(self, timer): timer.start_task("task-1.1") result = timer.end_task("task-1.1") assert "start_time" in result datetime.fromisoformat(result["start_time"]) def test_returns_end_time(self, timer): timer.start_task("task-1.1") result = timer.end_task("task-1.1") assert "end_time" in result datetime.fromisoformat(result["end_time"]) def test_returns_duration_seconds(self, timer): timer.start_task("task-1.1") result = timer.end_task("task-1.1") assert "duration_seconds" in result assert isinstance(result["duration_seconds"], (int, float)) assert result["duration_seconds"] >= 0 def test_returns_duration_human(self, timer): timer.start_task("task-1.1") result = timer.end_task("task-1.1") assert "duration_human" in result assert isinstance(result["duration_human"], str) assert len(result["duration_human"]) > 0 def test_task_status_updated_in_file(self, timer, tmp_path): timer.start_task("task-1.1") timer.end_task("task-1.1") timer_file = tmp_path / "memory" / "task-timers.json" data = json.loads(timer_file.read_text()) assert data["tasks"]["task-1.1"]["status"] == "completed" def test_done_event_file_created(self, timer, tmp_path): """end_task() 호출 시 .done 이벤트 파일이 생성되어야 함""" timer.start_task("task-1.1") timer.end_task("task-1.1") done_file = tmp_path / "memory" / "events" / "task-1.1.done" assert done_file.exists() def test_done_event_file_contains_valid_json(self, timer, tmp_path): timer.start_task("task-1.1") timer.end_task("task-1.1") done_file = tmp_path / "memory" / "events" / "task-1.1.done" event_data = json.loads(done_file.read_text()) assert event_data["task_id"] == "task-1.1" assert "end_time" in event_data assert "duration_seconds" in event_data def test_end_time_is_after_start_time(self, timer): timer.start_task("task-1.1") result = timer.end_task("task-1.1") start = datetime.fromisoformat(result["start_time"]) end = datetime.fromisoformat(result["end_time"]) assert end >= start # --------------------------------------------------------------------------- # 3. 존재하지 않는 task_id → status="error" # --------------------------------------------------------------------------- class TestEndTaskNotFound: """존재하지 않는 task_id로 end_task() 호출 시 에러 응답 검증""" def test_returns_status_error(self, timer): result = timer.end_task("task-nonexistent") assert result["status"] == "error" def test_error_response_contains_reason(self, timer): result = timer.end_task("task-nonexistent") assert "reason" in result assert "nonexistent" in result["reason"] or "not found" in result["reason"].lower() def test_no_done_file_created_on_error(self, timer, tmp_path): timer.end_task("task-ghost") done_file = tmp_path / "memory" / "events" / "task-ghost.done" assert not done_file.exists() # --------------------------------------------------------------------------- # 4. get_task_status() 조회 # --------------------------------------------------------------------------- class TestGetTaskStatus: """get_task_status()가 올바른 작업 상태를 반환하는지 확인""" def test_returns_none_for_nonexistent_task(self, timer): result = timer.get_task_status("task-nonexistent") assert result is None def test_returns_running_status(self, timer): timer.start_task("task-1.1") result = timer.get_task_status("task-1.1") assert result is not None assert result["status"] == "running" def test_returns_completed_status(self, timer): timer.start_task("task-1.1") timer.end_task("task-1.1") result = timer.get_task_status("task-1.1") assert result["status"] == "completed" def test_returns_task_data_dict(self, timer): timer.start_task("task-1.1", team_id="dev1-team", description="테스트") result = timer.get_task_status("task-1.1") assert result["task_id"] == "task-1.1" assert result["team_id"] == "dev1-team" assert result["description"] == "테스트" # --------------------------------------------------------------------------- # 5. list_tasks() 동작 검증 # --------------------------------------------------------------------------- class TestListTasks: """list_tasks()가 올바르게 작업 목록을 반환하는지 확인""" def test_empty_list_initially(self, timer): result = timer.list_tasks() assert result["total"] == 0 assert result["tasks"] == [] def test_total_count_after_start(self, timer): timer.start_task("task-1.1") timer.start_task("task-2.1") result = timer.list_tasks() assert result["total"] == 2 def test_list_all_tasks(self, timer): timer.start_task("task-1.1") timer.start_task("task-2.1") timer.end_task("task-1.1") result = timer.list_tasks() assert result["total"] == 2 def test_list_filter_by_running(self, timer): timer.start_task("task-1.1") timer.start_task("task-2.1") timer.end_task("task-1.1") result = timer.list_tasks(status="running") assert result["total"] == 1 assert result["tasks"][0]["task_id"] == "task-2.1" def test_list_filter_by_completed(self, timer): timer.start_task("task-1.1") timer.start_task("task-2.1") timer.end_task("task-1.1") result = timer.list_tasks(status="completed") assert result["total"] == 1 assert result["tasks"][0]["task_id"] == "task-1.1" def test_list_returns_dict_with_total_and_tasks_keys(self, timer): result = timer.list_tasks() assert "total" in result assert "tasks" in result def test_list_tasks_data_structure(self, timer): """각 task 항목에 필수 필드가 포함되어 있어야 함""" timer.start_task("task-1.1", team_id="dev1-team", description="테스트") result = timer.list_tasks() task = result["tasks"][0] assert "task_id" in task assert "status" in task assert "start_time" in task def test_list_with_nonexistent_status_returns_empty(self, timer): timer.start_task("task-1.1") result = timer.list_tasks(status="nonexistent") assert result["total"] == 0 assert result["tasks"] == [] # --------------------------------------------------------------------------- # 5. _format_duration() 단위 변환 검증 # --------------------------------------------------------------------------- class TestFormatDuration: """_format_duration()이 초/분/시간을 올바른 한국어 문자열로 변환하는지 확인""" @pytest.fixture(autouse=True) def _timer_instance(self, timer): self.timer = timer def test_seconds_under_60(self): assert self.timer._format_duration(30) == "30초" def test_seconds_zero(self): assert self.timer._format_duration(0) == "0초" def test_seconds_59(self): assert self.timer._format_duration(59) == "59초" def test_seconds_exactly_60_is_one_minute(self): result = self.timer._format_duration(60) assert "분" in result assert "1분" in result def test_minutes_and_seconds(self): result = self.timer._format_duration(90) # 1분 30초 assert "1분" in result assert "30초" in result def test_minutes_without_seconds(self): result = self.timer._format_duration(120) # 2분 0초 assert "2분" in result def test_just_under_one_hour(self): result = self.timer._format_duration(3599) # 59분 59초 assert "59분" in result assert "시간" not in result def test_exactly_one_hour(self): result = self.timer._format_duration(3600) assert "1시간" in result def test_one_hour_thirty_minutes(self): result = self.timer._format_duration(5400) # 1시간 30분 assert "1시간" in result assert "30분" in result def test_two_hours(self): result = self.timer._format_duration(7200) assert "2시간" in result def test_format_returns_string(self): assert isinstance(self.timer._format_duration(100), str) # --------------------------------------------------------------------------- # 6. add_log_entry() 일일 로그 파일 생성 확인 # --------------------------------------------------------------------------- class TestAddLogEntry: """add_log_entry()가 일일 로그 파일을 올바르게 생성/추가하는지 확인""" def test_returns_status_logged(self, timer): result = timer.add_log_entry("테스트 메시지") assert result["status"] == "logged" def test_returns_message(self, timer): result = timer.add_log_entry("테스트 메시지") assert result["message"] == "테스트 메시지" def test_returns_entry_type(self, timer): result = timer.add_log_entry("테스트", entry_type="note") assert result["type"] == "note" def test_returns_timestamp(self, timer): result = timer.add_log_entry("테스트") assert "timestamp" in result datetime.fromisoformat(result["timestamp"]) def test_creates_daily_log_file(self, timer, tmp_path): timer.add_log_entry("로그 항목 테스트") daily_dir = tmp_path / "memory" / "daily" log_files = list(daily_dir.glob("*.md")) assert len(log_files) == 1 def test_daily_log_file_name_is_today(self, timer, tmp_path): timer.add_log_entry("날짜 테스트") today_str = datetime.now().strftime("%Y-%m-%d") daily_dir = tmp_path / "memory" / "daily" expected_file = daily_dir / f"{today_str}.md" assert expected_file.exists() def test_log_file_contains_message(self, timer, tmp_path): timer.add_log_entry("고유한 메시지 내용 XYZ123") today_str = datetime.now().strftime("%Y-%m-%d") log_file = tmp_path / "memory" / "daily" / f"{today_str}.md" content = log_file.read_text(encoding="utf-8") assert "고유한 메시지 내용 XYZ123" in content def test_multiple_log_entries_appended(self, timer, tmp_path): timer.add_log_entry("첫 번째 메시지") timer.add_log_entry("두 번째 메시지") today_str = datetime.now().strftime("%Y-%m-%d") log_file = tmp_path / "memory" / "daily" / f"{today_str}.md" content = log_file.read_text(encoding="utf-8") assert "첫 번째 메시지" in content assert "두 번째 메시지" in content def test_dispatch_type_log_entry(self, timer): result = timer.add_log_entry("팀 위임 기록", entry_type="dispatch") assert result["status"] == "logged" assert result["type"] == "dispatch" def test_decision_type_log_entry(self, timer, tmp_path): timer.add_log_entry("중요 의사결정 내용", entry_type="decision") today_str = datetime.now().strftime("%Y-%m-%d") log_file = tmp_path / "memory" / "daily" / f"{today_str}.md" content = log_file.read_text(encoding="utf-8") assert "중요 의사결정 내용" in content def test_system_type_log_entry(self, timer, tmp_path): timer.add_log_entry("시스템 설정 변경", entry_type="system") today_str = datetime.now().strftime("%Y-%m-%d") log_file = tmp_path / "memory" / "daily" / f"{today_str}.md" content = log_file.read_text(encoding="utf-8") assert "시스템 설정 변경" in content def test_start_task_writes_to_daily_log(self, timer, tmp_path): """start_task()에 team_id 또는 description이 있으면 일일 로그에 기록됨""" timer.start_task("task-1.1", team_id="dev1-team", description="로그 테스트") daily_dir = tmp_path / "memory" / "daily" log_files = list(daily_dir.glob("*.md")) assert len(log_files) == 1 def test_end_task_writes_to_daily_log(self, timer, tmp_path): """end_task()도 team_id/description이 있으면 일일 로그에 완료 기록""" timer.start_task("task-1.1", team_id="dev1-team", description="완료 로그 테스트") timer.end_task("task-1.1") today_str = datetime.now().strftime("%Y-%m-%d") log_file = tmp_path / "memory" / "daily" / f"{today_str}.md" content = log_file.read_text(encoding="utf-8") assert "completed" in content or "완료" in content # --------------------------------------------------------------------------- # 8. CLI main() 함수 테스트 # --------------------------------------------------------------------------- _main_func = _task_timer_mod.main _original_init = TaskTimer.__init__ def _make_patched_init(tmp_path): """재귀 방지를 위해 원본 __init__을 캡처한 패치 생성""" def patched_init(self, workspace_path=None): _original_init(self, str(tmp_path)) return patched_init class TestCLIMain: """CLI main() 함수의 주요 명령어 테스트""" def test_start_command(self, tmp_path, monkeypatch, capsys): """start 명령어가 JSON 출력을 생성하는지 확인""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr( sys, "argv", ["task-timer.py", "start", "task-1.1", "--team", "dev1-team", "--desc", "CLI 테스트"] ) _main_func() captured = capsys.readouterr() output = json.loads(captured.out) assert output["status"] == "started" def test_end_command_after_start(self, tmp_path, monkeypatch, capsys): """start 후 end 명령어가 정상 동작하는지 확인""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "start", "task-2.1"]) _main_func() monkeypatch.setattr(sys, "argv", ["task-timer.py", "end", "task-2.1"]) _main_func() captured = capsys.readouterr() assert "completed" in captured.out def test_list_command(self, tmp_path, monkeypatch, capsys): """list 명령어가 작업 목록을 반환하는지 확인""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "list"]) _main_func() captured = capsys.readouterr() output = json.loads(captured.out) assert "total" in output assert "tasks" in output def test_status_command(self, tmp_path, monkeypatch, capsys): """status 명령어가 작업 상태를 반환하는지 확인""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "start", "task-3.1"]) _main_func() monkeypatch.setattr(sys, "argv", ["task-timer.py", "status", "task-3.1"]) _main_func() captured = capsys.readouterr() assert "running" in captured.out def test_status_not_found_exits(self, tmp_path, monkeypatch): """존재하지 않는 task status 조회 시 sys.exit(1)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "status", "task-999.9"]) with pytest.raises(SystemExit) as exc_info: _main_func() assert exc_info.value.code == 1 def test_log_command(self, tmp_path, monkeypatch, capsys): """log 명령어가 로그 항목을 생성하는지 확인""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "log", "테스트 로그 메시지", "--type", "note"]) _main_func() captured = capsys.readouterr() output = json.loads(captured.out) assert output["status"] == "logged" def test_no_args_exits(self, monkeypatch): """인자 없이 실행 시 sys.exit(1)""" monkeypatch.setattr(sys, "argv", ["task-timer.py"]) with pytest.raises(SystemExit) as exc_info: _main_func() assert exc_info.value.code == 1 def test_unknown_command_exits(self, monkeypatch): """알 수 없는 명령어 시 sys.exit(1)""" monkeypatch.setattr(sys, "argv", ["task-timer.py", "invalid_cmd"]) with pytest.raises(SystemExit) as exc_info: _main_func() assert exc_info.value.code == 1 def test_log_invalid_type_exits(self, tmp_path, monkeypatch): """잘못된 log type 시 sys.exit(1)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "log", "msg", "--type", "invalid_type"]) with pytest.raises(SystemExit) as exc_info: _main_func() assert exc_info.value.code == 1 def test_start_command_task_n_format(self, tmp_path, monkeypatch, capsys): """CLI에서 "start task-500" → started (task-N 형식 허용)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "start", "task-500"]) _main_func() captured = capsys.readouterr() output = json.loads(captured.out) assert output["status"] == "started" def test_end_command_fuzzy_match(self, tmp_path, monkeypatch, capsys): """start task-500.1 후 CLI에서 "end task-500" → completed""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "start", "task-500.1"]) _main_func() monkeypatch.setattr(sys, "argv", ["task-timer.py", "end", "task-500"]) _main_func() captured = capsys.readouterr() assert "completed" in captured.out # --------------------------------------------------------------------------- # 9. _load_timers() 에러 핸들링 (lines 55-57) # --------------------------------------------------------------------------- class TestLoadTimersError: """_load_timers() 에러 핸들링 테스트 (lines 55-57)""" def test_corrupted_json_returns_empty(self, tmp_path): """깨진 JSON 파일 → 빈 tasks 반환""" (tmp_path / "memory").mkdir(parents=True, exist_ok=True) timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text("NOT VALID JSON {{{", encoding="utf-8") timer = TaskTimer(workspace_path=str(tmp_path)) assert timer.timers == {"tasks": {}} def test_unreadable_file_returns_empty(self, tmp_path): """읽기 권한 없는 파일 → 빈 tasks 반환""" import os (tmp_path / "memory").mkdir(parents=True, exist_ok=True) timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text('{"tasks": {}}', encoding="utf-8") os.chmod(str(timer_file), 0o000) try: timer = TaskTimer(workspace_path=str(tmp_path)) assert timer.timers == {"tasks": {}} finally: os.chmod(str(timer_file), 0o644) # --------------------------------------------------------------------------- # 10. _save_timers() 에러 핸들링 (lines 72-73) # --------------------------------------------------------------------------- class TestSaveTimersError: """_save_timers() 에러 핸들링 테스트 (lines 72-73)""" def test_save_to_readonly_dir(self, tmp_path): """읽기 전용 디렉토리에 저장 시도 → 에러 무시하고 계속""" import os (tmp_path / "memory").mkdir(parents=True, exist_ok=True) timer = TaskTimer(workspace_path=str(tmp_path)) # memory 디렉토리를 읽기 전용으로 설정 os.chmod(str(tmp_path / "memory"), 0o555) try: # 저장이 실패해도 예외가 발생하지 않아야 함 timer.timers["tasks"]["test"] = {"status": "test"} timer._save_timers() # 에러 발생하지만 내부에서 처리 finally: os.chmod(str(tmp_path / "memory"), 0o755) # --------------------------------------------------------------------------- # 11. _update_pipeline_status() 에지 케이스 (lines 186-187, 231-233) # --------------------------------------------------------------------------- class TestUpdatePipelineStatusEdgeCases: """_update_pipeline_status() 에지 케이스 (lines 186-187, 231-233)""" def test_corrupted_pipeline_json(self, tmp_path): """pipeline-status.json이 깨진 JSON → 빈 dict로 초기화 (lines 186-187)""" (tmp_path / "memory").mkdir(parents=True, exist_ok=True) pipeline_file = tmp_path / "memory" / "pipeline-status.json" pipeline_file.write_text("INVALID{{{JSON", encoding="utf-8") timer = TaskTimer(workspace_path=str(tmp_path)) # start_task 내부에서 _update_pipeline_status를 호출 result = timer.start_task("task-5.1", team_id="dev1-team") assert result["status"] == "started" # pipeline 파일이 올바른 JSON으로 덮어써졌는지 확인 data = json.loads(pipeline_file.read_text()) assert "active_tasks" in data def test_pipeline_update_exception_ignored(self, tmp_path): """pipeline-status 업데이트 실패해도 task-timer 기능은 계속 (lines 231-233)""" from unittest.mock import MagicMock, patch (tmp_path / "memory").mkdir(parents=True, exist_ok=True) timer = TaskTimer(workspace_path=str(tmp_path)) # pipeline_status_file.parent.mkdir()에서 예외를 강제로 발생시킴 mock_path = MagicMock() mock_path.parent.mkdir.side_effect = OSError("강제 오류") mock_path.exists.return_value = False with patch.object(timer, "pipeline_status_file", mock_path): # Exception이 발생해도 pass 처리 → 예외 없이 통과해야 함 timer._update_pipeline_status("start", {"task_id": "test-task"}) # 예외 없이 통과했으면 성공 (lines 231-233 커버) # --------------------------------------------------------------------------- # 12. _append_to_section() 상세 테스트 (lines 313-346) # --------------------------------------------------------------------------- class TestAppendToSection: """_append_to_section() 함수 상세 테스트 (lines 313-346)""" def test_section_exists_insert_before_next_section(self, timer, tmp_path): """기존 섹션이 있고 다음 섹션이 있을 때, 다음 섹션 직전에 삽입 (lines 324-340)""" today_str = datetime.now().strftime("%Y-%m-%d") daily_dir = tmp_path / "memory" / "daily" daily_dir.mkdir(parents=True, exist_ok=True) log_file = daily_dir / f"{today_str}.md" content = f"# {today_str} 업무일지\n\n## 완료된 작업\n- 기존 항목\n\n## 의사결정\n- 기존 의사결정\n\n## 시스템 변경\n- 기존 시스템\n" log_file.write_text(content, encoding="utf-8") timer._append_to_section("- 새 의사결정 항목", "의사결정") result = log_file.read_text(encoding="utf-8") assert "새 의사결정 항목" in result # 시스템 변경 섹션보다 앞에 삽입되어야 함 idx_new = result.index("새 의사결정 항목") idx_system = result.index("## 시스템 변경") assert idx_new < idx_system def test_section_exists_no_next_section(self, timer, tmp_path): """기존 섹션이 있고 다음 섹션이 없을 때, 파일 끝에 추가 (lines 337-338)""" today_str = datetime.now().strftime("%Y-%m-%d") daily_dir = tmp_path / "memory" / "daily" daily_dir.mkdir(parents=True, exist_ok=True) log_file = daily_dir / f"{today_str}.md" content = f"# {today_str} 업무일지\n\n## 완료된 작업\n- 기존 항목\n\n## 의사결정\n- 기존 결정" log_file.write_text(content, encoding="utf-8") timer._append_to_section("- 새 의사결정", "의사결정") result = log_file.read_text(encoding="utf-8") assert "새 의사결정" in result def test_new_section_created_when_missing(self, timer, tmp_path): """섹션이 없으면 파일 끝에 새 섹션 추가 (lines 343-346)""" today_str = datetime.now().strftime("%Y-%m-%d") daily_dir = tmp_path / "memory" / "daily" daily_dir.mkdir(parents=True, exist_ok=True) log_file = daily_dir / f"{today_str}.md" content = f"# {today_str} 업무일지\n\n## 완료된 작업\n- 기존 항목" log_file.write_text(content, encoding="utf-8") timer._append_to_section("- 아키텍처 논의 내용", "아키텍처 논의") result = log_file.read_text(encoding="utf-8") assert "## 아키텍처 논의" in result assert "아키텍처 논의 내용" in result def test_content_without_newline_ending(self, timer, tmp_path): """content가 \\n으로 끝나지 않을 때 (line 344)""" today_str = datetime.now().strftime("%Y-%m-%d") daily_dir = tmp_path / "memory" / "daily" daily_dir.mkdir(parents=True, exist_ok=True) log_file = daily_dir / f"{today_str}.md" # \n으로 끝나지 않는 content content = f"# {today_str} 업무일지\n\n## 완료된 작업\n- 기존 항목" assert not content.endswith("\n") log_file.write_text(content, encoding="utf-8") timer._append_to_section("- 새 시스템 변경", "시스템 변경") result = log_file.read_text(encoding="utf-8") assert "## 시스템 변경" in result def test_wrong_date_header_gets_prepended(self, timer, tmp_path): """날짜 헤더가 맞지 않을 때 prepend (line 318)""" today_str = datetime.now().strftime("%Y-%m-%d") daily_dir = tmp_path / "memory" / "daily" daily_dir.mkdir(parents=True, exist_ok=True) log_file = daily_dir / f"{today_str}.md" # 잘못된 날짜 헤더 content = "# 2020-01-01 업무일지\n\n## 완료된 작업\n- 옛날 항목" log_file.write_text(content, encoding="utf-8") timer._append_to_section("- 새 항목", "위임 기록") result = log_file.read_text(encoding="utf-8") assert f"# {today_str}" in result # --------------------------------------------------------------------------- # 13. CLI main() 추가 에지 케이스 (lines 396-397, 414-423, 430-431, 439-440, 458-459) # --------------------------------------------------------------------------- class TestCLIMainEdgeCases: """CLI main() 추가 에지 케이스 테스트""" def test_start_without_task_id_exits(self, tmp_path, monkeypatch): """start 명령어에 task_id 누락 시 sys.exit(1) (lines 396-397)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "start"]) with pytest.raises(SystemExit) as exc_info: _main_func() assert exc_info.value.code == 1 def test_end_without_task_id_exits(self, tmp_path, monkeypatch): """end 명령어에 task_id 누락 시 sys.exit(1) (lines 430-431)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "end"]) with pytest.raises(SystemExit) as exc_info: _main_func() assert exc_info.value.code == 1 def test_status_without_task_id_exits(self, tmp_path, monkeypatch): """status 명령어에 task_id 누락 시 sys.exit(1) (lines 439-440)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "status"]) with pytest.raises(SystemExit) as exc_info: _main_func() assert exc_info.value.code == 1 def test_log_without_message_exits(self, tmp_path, monkeypatch): """log 명령어에 메시지 누락 시 sys.exit(1) (lines 458-459)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "log"]) with pytest.raises(SystemExit) as exc_info: _main_func() assert exc_info.value.code == 1 def test_start_with_positional_args(self, tmp_path, monkeypatch, capsys): """start 명령어에 위치 인자로 team_id, description 전달 (lines 414-423)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr(sys, "argv", ["task-timer.py", "start", "task-10.1", "dev1-team", "위치인자 설명"]) _main_func() captured = capsys.readouterr() output = json.loads(captured.out) assert output["status"] == "started" def test_start_with_project_flag(self, tmp_path, monkeypatch, capsys): """start 명령어에 --project 플래그 전달 (lines 414-416)""" monkeypatch.setattr(_task_timer_mod.TaskTimer, "__init__", _make_patched_init(tmp_path)) monkeypatch.setattr( sys, "argv", ["task-timer.py", "start", "task-11.1", "--team", "dev1-team", "--project", "myproj"] ) _main_func() captured = capsys.readouterr() output = json.loads(captured.out) assert output["status"] == "started" # --------------------------------------------------------------------------- # 14. task_id / team_id 형식 검증 # --------------------------------------------------------------------------- # 모듈 레벨에서 validate 함수 가져오기 validate_task_id = _task_timer_mod.validate_task_id validate_team_id = _task_timer_mod.validate_team_id class TestTaskIdValidation: """task_id 형식 검증 테스트""" def test_valid_task_id_simple(self): assert validate_task_id("task-1.1") is True def test_valid_task_id_large_numbers(self): assert validate_task_id("task-401.1") is True def test_valid_task_id_multi_digit_sub(self): assert validate_task_id("task-12.3") is True def test_invalid_task_id_no_prefix(self): assert validate_task_id("401.1") is False def test_invalid_task_id_wrong_prefix(self): assert validate_task_id("odin-2") is False def test_valid_task_id_no_dot(self): assert validate_task_id("task-401") is True def test_valid_task_id_large_no_dot(self): assert validate_task_id("task-1000") is True def test_start_task_accepts_task_n_format(self, timer): result = timer.start_task("task-500") assert result["status"] == "started" def test_invalid_task_id_alpha(self): assert validate_task_id("task-abc.1") is False def test_invalid_task_id_empty(self): assert validate_task_id("") is False def test_invalid_task_id_spaces(self): assert validate_task_id("task- 1.1") is False def test_valid_task_id_phase(self): assert validate_task_id("task-1845_3.3") is True def test_valid_task_id_parallel(self): assert validate_task_id("task-1845_a") is True def test_valid_task_id_phase_parallel(self): assert validate_task_id("task-1845_3.2_b") is True def test_valid_task_id_retry(self): assert validate_task_id("task-1845+1") is True def test_valid_task_id_full_v2(self): assert validate_task_id("task-1845_3.2_b+1") is True def test_valid_task_id_phase_retry(self): assert validate_task_id("task-1845_3.3+2") is True def test_valid_task_id_parallel_retry(self): assert validate_task_id("task-1845_a+1") is True def test_invalid_task_id_uppercase_parallel(self): assert validate_task_id("task-1845_A") is False def test_invalid_task_id_bad_phase(self): assert validate_task_id("task-1845_3") is False def test_invalid_task_id_double_underscore(self): assert validate_task_id("task-1845__3.3") is False def test_start_task_rejects_invalid_task_id(self, timer): result = timer.start_task("invalid-id") assert result["status"] == "error" assert "Invalid task_id" in result["reason"] def test_start_task_accepts_valid_task_id(self, timer): result = timer.start_task("task-100.1") assert result["status"] == "started" class TestFuzzyMatchEnd: """end_task() fuzzy match 테스트""" def test_end_fuzzy_match_single_running(self, timer): """task-1.1을 start한 뒤 end_task("task-1")로 종료 → completed""" timer.start_task("task-1.1") result = timer.end_task("task-1") assert result["status"] == "completed" def test_end_fuzzy_match_multiple_running_picks_latest(self, timer): """task-1.1과 task-1.2를 start한 뒤 end_task("task-1") → task-1.2가 종료됨 (최신)""" timer.start_task("task-1.1") timer.start_task("task-1.2") result = timer.end_task("task-1") assert result["status"] == "completed" assert result["task_id"] == "task-1.2" def test_end_exact_match_preferred_over_fuzzy(self, timer): """task-1과 task-1.1 둘 다 start한 뒤 end_task("task-1") → task-1이 종료됨 (정확 매칭 우선)""" timer.start_task("task-1") timer.start_task("task-1.1") result = timer.end_task("task-1") assert result["status"] == "completed" assert result["task_id"] == "task-1" def test_end_fuzzy_no_running_returns_error(self, timer): """task-1.1을 start + end한 뒤 (completed 상태) end_task("task-1") → error""" timer.start_task("task-1.1") timer.end_task("task-1.1") result = timer.end_task("task-1") assert result["status"] == "error" def test_end_exact_still_works(self, timer): """task-1.1 start 후 end_task("task-1.1") → completed (기존 동작 유지)""" timer.start_task("task-1.1") result = timer.end_task("task-1.1") assert result["status"] == "completed" class TestFuzzyMatchStatus: """get_task_status() fuzzy match 테스트""" def test_status_fuzzy_match_single(self, timer): """task-1.1 start 후 get_task_status("task-1") → running 상태 반환""" timer.start_task("task-1.1") result = timer.get_task_status("task-1") assert result is not None assert result["status"] == "running" def test_status_exact_match_preferred(self, timer): """task-1, task-1.1 둘 다 start 후 get_task_status("task-1") → task-1 반환""" timer.start_task("task-1") timer.start_task("task-1.1") result = timer.get_task_status("task-1") assert result is not None assert result["task_id"] == "task-1" def test_status_no_match_returns_none(self, timer): """get_task_status("task-999") → None""" result = timer.get_task_status("task-999") assert result is None class TestTeamIdValidation: """team_id 형식 검증 테스트""" def test_valid_dev1_team(self): assert validate_team_id("dev1-team") is True def test_valid_dev2_team(self): assert validate_team_id("dev2-team") is True def test_valid_dev3_team(self): assert validate_team_id("dev3-team") is True def test_valid_anu_direct(self): assert validate_team_id("anu-direct") is True def test_valid_empty_string(self): assert validate_team_id("") is True def test_invalid_odin_2(self): assert validate_team_id("odin-2") is False def test_valid_dev4_team(self): """dev4-team은 유효한 팀 ID""" assert validate_team_id("dev4-team") is True def test_invalid_random(self): assert validate_team_id("random-team") is False def test_start_task_rejects_invalid_team_id(self, timer): result = timer.start_task("task-100.1", team_id="odin-2") assert result["status"] == "error" assert "Invalid team_id" in result["reason"] def test_start_task_accepts_empty_team_id(self, timer): result = timer.start_task("task-100.1", team_id="") assert result["status"] == "started" def test_start_task_accepts_valid_team_id(self, timer): result = timer.start_task("task-100.1", team_id="dev2-team") assert result["status"] == "started" def test_valid_design_team(self): """design 팀 ID가 유효함을 검증 (논리적 팀)""" assert validate_team_id("design") is True def test_valid_publishing_team(self): """publishing 팀 ID가 유효함을 검증 (논리적 팀)""" assert validate_team_id("publishing") is True def test_valid_composite_team(self): """composite 팀 ID가 유효함을 검증 (복합업무)""" assert validate_team_id("composite") is True def test_start_task_accepts_design_team(self, timer): """start_task()가 design 팀 ID를 수락하는지 검증""" result = timer.start_task("task-100.2", team_id="design") assert result["status"] == "started" def test_start_task_accepts_publishing_team(self, timer): """start_task()가 publishing 팀 ID를 수락하는지 검증""" result = timer.start_task("task-100.3", team_id="publishing") assert result["status"] == "started" def test_start_task_accepts_composite_team(self, timer): """start_task()가 composite 팀 ID를 수락하는지 검증""" result = timer.start_task("task-100.4", team_id="composite") assert result["status"] == "started" class TestAtomicSave: """_save_timers() 원자적 쓰기 테스트""" def test_save_creates_valid_json(self, timer, tmp_path): """저장 후 파일이 유효한 JSON인지 확인""" timer.start_task("task-1.1", team_id="dev1-team", description="원자적 저장 테스트") timer_file = tmp_path / "memory" / "task-timers.json" data = json.loads(timer_file.read_text(encoding="utf-8")) assert "task-1.1" in data["tasks"] assert data["tasks"]["task-1.1"]["status"] == "running" def test_save_does_not_leave_tmp_files(self, timer, tmp_path): """저장 후 .tmp 파일이 남아있지 않아야 함""" timer.start_task("task-2.1") memory_dir = tmp_path / "memory" tmp_files = list(memory_dir.glob("*.tmp")) assert len(tmp_files) == 0, f"임시 파일이 남아있음: {tmp_files}" def test_concurrent_save_preserves_data(self, tmp_path): """두 TaskTimer 인스턴스가 순차적으로 저장해도 데이터가 유실되지 않음""" (tmp_path / "memory").mkdir(parents=True, exist_ok=True) timer1 = TaskTimer(workspace_path=str(tmp_path)) timer1.start_task("task-1.1", team_id="dev1-team") timer2 = TaskTimer(workspace_path=str(tmp_path)) timer2.start_task("task-2.1", team_id="dev2-team") timer_file = tmp_path / "memory" / "task-timers.json" data = json.loads(timer_file.read_text(encoding="utf-8")) # timer2가 timer1의 데이터를 로드한 상태에서 시작하므로 둘 다 존재 assert "task-1.1" in data["tasks"] assert "task-2.1" in data["tasks"] def test_save_uses_lock_file(self, timer, tmp_path): """저장 시 .task-timers.lock 파일이 생성됨""" timer.start_task("task-3.1") lock_file = tmp_path / "memory" / ".task-timers.lock" assert lock_file.exists(), ".task-timers.lock 파일이 생성되어야 함" class TestStaleGuard: """stale 상태 task 재시작 방지 테스트 (task-486.1)""" def test_start_task_rejects_stale(self, timer): """stale 상태의 task에 start_task 호출 시 error 반환""" # 먼저 task를 시작 timer.start_task("task-99.1", "dev1-team", "테스트 작업") # 수동으로 stale 상태로 변경 timer.timers["tasks"]["task-99.1"]["status"] = "stale" timer.timers["tasks"]["task-99.1"]["stale_reason"] = "timeout_running" timer._save_timers() # stale 상태에서 재시작 시도 → 거부되어야 함 result = timer.start_task("task-99.1", "dev1-team", "재시작 시도") assert result["status"] == "error" assert "stale" in result["reason"] def test_start_task_stale_preserves_original_data(self, timer): """stale 거부 후 기존 task 데이터가 변경되지 않아야 함""" timer.start_task("task-99.1", "dev1-team", "원래 작업") timer.timers["tasks"]["task-99.1"]["status"] = "stale" timer._save_timers() # 재시작 시도 timer.start_task("task-99.1", "dev1-team", "덮어쓰기 시도") # 원래 데이터 보존 확인 task_data = timer.timers["tasks"]["task-99.1"] assert task_data["status"] == "stale" assert task_data["description"] == "원래 작업" # --------------------------------------------------------------------------- # 횡단조직 추적 테스트 (cross-start / cross-end) # --------------------------------------------------------------------------- class TestCrossStart: """cross_start 메서드 테스트""" def test_cross_start_valid_agent(self, timer): """유효한 agent로 cross_start 성공""" result = timer.cross_start("loki", "task-548.1", "보안 취약점 리뷰", "dev1-team") assert result["status"] == "started" assert result["agent"] == "loki" assert result["task_id"] == "task-548.1" def test_cross_start_invalid_agent(self, timer): """잘못된 agent_name → 에러""" result = timer.cross_start("unknown-agent", "task-1.1", "설명", "dev1-team") assert result["status"] == "error" assert "agent" in result["reason"].lower() or "unknown" in result["reason"].lower() def test_cross_start_invalid_task_id(self, timer): """잘못된 task_id → 에러""" result = timer.cross_start("loki", "invalid-task", "설명", "dev1-team") assert result["status"] == "error" assert "task_id" in result["reason"].lower() or "invalid" in result["reason"].lower() def test_cross_start_overwrite_active(self, timer): """이미 active인 agent에 다시 cross_start → 덮어쓰기""" timer.cross_start("loki", "task-1.1", "첫 번째 작업", "dev1-team") result = timer.cross_start("loki", "task-2.1", "두 번째 작업", "dev2-team") assert result["status"] == "started" assert result["task_id"] == "task-2.1" # 파일에서도 최신 상태로 덮어써졌는지 확인 cross_data = timer._load_cross_status() assert cross_data["cross_functional"]["loki"]["task_id"] == "task-2.1" class TestCrossEnd: """cross_end 메서드 테스트""" def test_cross_end_active_agent(self, timer): """active 상태 agent 종료 → idle""" timer.cross_start("venus", "task-1.1", "디자인 리뷰", "dev1-team") result = timer.cross_end("venus") assert result["status"] == "ended" assert result["agent"] == "venus" cross_data = timer._load_cross_status() assert cross_data["cross_functional"]["venus"]["status"] == "idle" def test_cross_end_idle_agent(self, timer): """이미 idle인 agent → 멱등적 성공""" result = timer.cross_end("maat") assert result["status"] == "ended" assert result["agent"] == "maat" def test_cross_end_invalid_agent(self, timer): """잘못된 agent_name → 에러""" result = timer.cross_end("nonexistent-agent") assert result["status"] == "error" class TestCrossFunctionalFile: """cross-functional-status.json 파일 IO 테스트""" def test_cross_start_creates_file(self, timer, tmp_path): """cross_start가 파일을 생성""" timer.cross_start("janus", "task-1.1", "DevOps 배포", "dev1-team") cross_file = tmp_path / "memory" / "cross-functional-status.json" assert cross_file.exists() def test_cross_status_persists(self, timer, tmp_path): """파일에 저장된 상태가 새 인스턴스에서 읽힘""" timer.cross_start("loki", "task-100.1", "보안 감사", "dev1-team") # 새 인스턴스로 읽기 new_timer = TaskTimer(workspace_path=str(tmp_path)) cross_data = new_timer._load_cross_status() assert cross_data["cross_functional"]["loki"]["status"] == "active" assert cross_data["cross_functional"]["loki"]["task_id"] == "task-100.1"