""" test_cross_status_recovery.py — TaskTimer의 _load_cross_status() 파일 복구 테스트 테스트 항목: 1. 손상된 JSON 파일을 생성 → _load_cross_status() 호출 → 정상 기본값 반환 확인 2. 호출 후 디스크 파일이 정상 JSON으로 복구되었는지 확인 3. 복구된 파일을 다시 로드하면 정상 기본값이 반환되는지 확인 """ import json import sys from pathlib import Path import pytest # task-timer 모듈 경로 추가 sys.path.insert(0, "/home/jay/workspace/memory") # task-timer.py를 동적으로 임포트 import importlib.util spec = importlib.util.spec_from_file_location("task_timer", "/home/jay/workspace/memory/task-timer.py") assert spec is not None and spec.loader is not None task_timer_module = importlib.util.module_from_spec(spec) sys.modules["task_timer"] = task_timer_module spec.loader.exec_module(task_timer_module) TaskTimer = task_timer_module.TaskTimer @pytest.fixture def tmp_workspace(tmp_path: Path) -> Path: """임시 workspace 생성""" memory_dir = tmp_path / "memory" memory_dir.mkdir(parents=True, exist_ok=True) return tmp_path class TestCrossStatusRecovery: """_load_cross_status() 파일 복구 기능 테스트""" def test_corrupted_json_file_recovery(self, tmp_workspace: Path) -> None: """ 손상된 JSON 파일 복구 테스트: 1. 손상된 JSON 파일 생성 2. _load_cross_status() 호출 3. 정상 기본값 반환 확인 4. 디스크 파일이 정상 JSON으로 복구됨 확인 """ # 손상된 JSON 파일 생성 cross_functional_file = tmp_workspace / "memory" / "cross-functional-status.json" cross_functional_file.write_text('{"invalid": json}', encoding="utf-8") # TaskTimer 인스턴스 생성 timer = TaskTimer(workspace_path=str(tmp_workspace)) # _load_cross_status() 호출 result = timer._load_cross_status() # 정상 기본값 반환 확인 assert result is not None assert "cross_functional" in result assert "loki" in result["cross_functional"] assert "venus" in result["cross_functional"] assert "maat" in result["cross_functional"] assert "janus" in result["cross_functional"] # 각 에이전트의 기본 상태 확인 for agent in ["loki", "venus", "maat", "janus"]: assert result["cross_functional"][agent]["status"] == "idle" # 디스크 파일이 정상 JSON으로 복구됨 확인 assert cross_functional_file.exists() recovered_content = cross_functional_file.read_text(encoding="utf-8") recovered_data = json.loads(recovered_content) assert "cross_functional" in recovered_data assert "loki" in recovered_data["cross_functional"] assert recovered_data["cross_functional"]["loki"]["status"] == "idle" def test_recovery_persists_across_reloads(self, tmp_workspace: Path) -> None: """ 복구된 파일이 다시 로드해도 유지되는지 확인: 1. 손상된 JSON 파일로 첫 번째 로드 2. 복구 확인 3. 새로운 TaskTimer 인스턴스로 재로드 4. 정상 데이터 반환 확인 """ # 손상된 JSON 파일 생성 cross_functional_file = tmp_workspace / "memory" / "cross-functional-status.json" cross_functional_file.write_text("corrupted json content", encoding="utf-8") # 첫 번째 로드 (복구) timer1 = TaskTimer(workspace_path=str(tmp_workspace)) result1 = timer1._load_cross_status() assert result1 is not None assert "cross_functional" in result1 # 두 번째 로드 (복구된 파일을 새 인스턴스로 로드) timer2 = TaskTimer(workspace_path=str(tmp_workspace)) result2 = timer2._load_cross_status() assert result2 is not None assert "cross_functional" in result2 # 두 번의 로드 결과가 일치 확인 assert result1 == result2 def test_partial_corrupted_json_recovery(self, tmp_workspace: Path) -> None: """ 부분적으로 손상된 JSON 파일 복구: 1. 부분적으로 손상된 JSON 파일 생성 2. _load_cross_status() 호출 3. 정상 기본값 반환 확인 """ cross_functional_file = tmp_workspace / "memory" / "cross-functional-status.json" # 부분적으로 손상된 JSON (닫히지 않은 괄호) cross_functional_file.write_text('{"cross_functional": {', encoding="utf-8") timer = TaskTimer(workspace_path=str(tmp_workspace)) result = timer._load_cross_status() # 정상 기본값 반환 확인 assert result is not None assert "cross_functional" in result assert all(agent in result["cross_functional"] for agent in ["loki", "venus", "maat", "janus"]) def test_empty_file_recovery(self, tmp_workspace: Path) -> None: """ 빈 파일 복구: 1. 빈 파일 생성 2. _load_cross_status() 호출 3. 정상 기본값 반환 확인 """ cross_functional_file = tmp_workspace / "memory" / "cross-functional-status.json" cross_functional_file.write_text("", encoding="utf-8") timer = TaskTimer(workspace_path=str(tmp_workspace)) result = timer._load_cross_status() # 정상 기본값 반환 확인 assert result is not None assert "cross_functional" in result for agent in ["loki", "venus", "maat", "janus"]: assert result["cross_functional"][agent]["status"] == "idle" def test_valid_json_not_overwritten(self, tmp_workspace: Path) -> None: """ 정상 JSON 파일은 덮어쓰지 않음 확인: 1. 정상 JSON 파일 생성 2. _load_cross_status() 호출 3. 파일 내용이 유지되는지 확인 """ cross_functional_file = tmp_workspace / "memory" / "cross-functional-status.json" original_data = { "cross_functional": { "loki": {"status": "active", "task_id": "task-100"}, "venus": {"status": "idle"}, "maat": {"status": "idle"}, "janus": {"status": "idle"}, } } cross_functional_file.write_text(json.dumps(original_data), encoding="utf-8") timer = TaskTimer(workspace_path=str(tmp_workspace)) result = timer._load_cross_status() # 정상 데이터 로드 확인 assert result == original_data assert result["cross_functional"]["loki"]["status"] == "active" assert result["cross_functional"]["loki"]["task_id"] == "task-100" def test_nonexistent_file_creation(self, tmp_workspace: Path) -> None: """ 파일이 없으면 생성 확인: 1. 파일 없음 2. _load_cross_status() 호출 3. 정상 파일이 생성됨 확인 """ cross_functional_file = tmp_workspace / "memory" / "cross-functional-status.json" # 파일이 없음 확인 assert not cross_functional_file.exists() timer = TaskTimer(workspace_path=str(tmp_workspace)) result = timer._load_cross_status() # 정상 기본값 반환 확인 assert result is not None assert "cross_functional" in result # 파일이 생성됨 확인 assert cross_functional_file.exists() saved_data = json.loads(cross_functional_file.read_text(encoding="utf-8")) assert saved_data == result class TestCrossStatusIntegration: """cross_start/cross_end와 함께 복구 기능 동작 확인""" def test_cross_start_after_recovery(self, tmp_workspace: Path) -> None: """ 파일 복구 후 cross_start 정상 동작 확인: 1. 손상된 파일 생성 2. _load_cross_status() 복구 3. cross_start 호출 4. 에이전트 상태 확인 """ cross_functional_file = tmp_workspace / "memory" / "cross-functional-status.json" cross_functional_file.write_text("invalid json", encoding="utf-8") timer = TaskTimer(workspace_path=str(tmp_workspace)) # 복구 _ = timer._load_cross_status() # cross_start 호출 result = timer.cross_start("loki", "task-100", "Test task") assert result["status"] == "started" assert result["agent"] == "loki" assert result["task_id"] == "task-100" # 파일에 정상 저장됨 확인 saved_data = json.loads(cross_functional_file.read_text(encoding="utf-8")) assert saved_data["cross_functional"]["loki"]["status"] == "active" assert saved_data["cross_functional"]["loki"]["task_id"] == "task-100" def test_cross_end_after_recovery(self, tmp_workspace: Path) -> None: """ 파일 복구 후 cross_end 정상 동작 확인 """ cross_functional_file = tmp_workspace / "memory" / "cross-functional-status.json" cross_functional_file.write_text("corrupted", encoding="utf-8") timer = TaskTimer(workspace_path=str(tmp_workspace)) # 복구 _ = timer._load_cross_status() # cross_start timer.cross_start("venus", "task-200", "Another task") # cross_end result = timer.cross_end("venus") assert result["status"] == "ended" assert result["agent"] == "venus" # 파일에 idle 상태로 저장됨 확인 saved_data = json.loads(cross_functional_file.read_text(encoding="utf-8")) assert saved_data["cross_functional"]["venus"]["status"] == "idle"