""" test_dispatch.py dispatch.py 단위 테스트 (아르고스 작성) 테스트 항목: - generate_task_id(): 유니크한 ID 생성 (tmp_path + monkeypatch로 WORKSPACE 격리) - build_prompt(): 올바른 팀 정보 포함 여부 확인 - TEAM_BOT 매핑 정확성 검증 - 잘못된 팀 ID 전달 시 에러 처리 (sys.exit 또는 ValueError) """ import json import os import sys import types from datetime import datetime from pathlib import Path from unittest.mock import MagicMock, patch import pytest _TEST_CHAT_ID = os.environ.get("COKACDIR_CHAT_ID", "6937032012") # --------------------------------------------------------------------------- # 헬퍼: organization-structure.json에서 동적 로드 # --------------------------------------------------------------------------- def _load_dev_team_ids_from_org() -> set: """organization-structure.json에서 dev 팀 ID 목록을 로드한다.""" workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) org_path = workspace / "memory" / "organization-structure.json" if not org_path.exists(): return set() with open(org_path, encoding="utf-8") as f: org = json.load(f) dev_team_ids = set() # dev 팀들은 development-office의 sub_teams 하위에 sub_team_id로 저장됨 for team in org.get("structure", {}).get("columns", {}).get("teams", []): if team.get("team_id") == "development-office": for sub in team.get("sub_teams", []): tid = sub.get("sub_team_id", "") if tid.startswith("dev") and tid.endswith("-team") and sub.get("status") != "planned": dev_team_ids.add(tid) return dev_team_ids def _count_all_team_info_from_org() -> int: """organization-structure.json에서 전체 팀(dev + logical) 수를 계산한다.""" workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) org_path = workspace / "memory" / "organization-structure.json" if not org_path.exists(): return -1 with open(org_path, encoding="utf-8") as f: org = json.load(f) count = 0 for team in org.get("structure", {}).get("columns", {}).get("teams", []): if team.get("status") == "planned": continue tid = team.get("team_id", "") if tid == "development-office": # dev 팀들은 sub_teams에서 카운트 for sub in team.get("sub_teams", []): if sub.get("status") != "planned": count += 1 elif tid in ("marketing-team", "consulting-team", "publishing-team", "design-team", "content-team"): # TEAM_INFO에는 marketing, consulting, publishing, design 키로 포함됨 count += 1 return count # --------------------------------------------------------------------------- # 헬퍼: dispatch 모듈을 격리된 WORKSPACE로 재로드 # --------------------------------------------------------------------------- def _load_dispatch_with_workspace(tmp_path: Path) -> types.ModuleType: """dispatch 모듈을 tmp_path를 WORKSPACE로 설정하여 로드한다. dispatch.py는 모듈 최상단에서 WORKSPACE를 __file__ 기준으로 결정하므로 sys.modules에서 제거 후 WORKSPACE 패치를 적용한다. """ workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) # sys.path에 workspace가 없으면 추가 if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) # prompts 패키지를 먼저 로드 (dispatch가 의존) import prompts.team_prompts # noqa: F401 # dispatch 모듈을 새로 로드 for mod_name in list(sys.modules.keys()): if mod_name == "dispatch": del sys.modules[mod_name] import dispatch as _dispatch # WORKSPACE를 tmp_path로 교체 _dispatch.WORKSPACE = tmp_path return _dispatch # --------------------------------------------------------------------------- # fixture: tmp_path 기반 dispatch 모듈 # --------------------------------------------------------------------------- @pytest.fixture() def dispatch_mod(tmp_path): """격리된 WORKSPACE를 사용하는 dispatch 모듈 반환""" import sys as _sys # memory 디렉토리 미리 생성 (tmp_path / "memory").mkdir(parents=True, exist_ok=True) (tmp_path / "memory" / "tasks").mkdir(parents=True, exist_ok=True) real_workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) # 기존 dispatch 모듈을 보존 (다른 테스트의 함수 __globals__ 보호) _original_dispatch = _sys.modules.get("dispatch") mod = _load_dispatch_with_workspace(tmp_path) # wake-up 로직이 테스트를 느리게 하지 않도록 _check_bot_process를 항상 True로 패치 mod._check_bot_process = lambda key: True # type: ignore[attr-defined] yield mod # 테스트 후 WORKSPACE를 실제 경로로 복원하여 다른 테스트 격리 mod.WORKSPACE = real_workspace # type: ignore[attr-defined] # sys.modules["dispatch"]를 원래 모듈로 복원 if _original_dispatch is not None: _sys.modules["dispatch"] = _original_dispatch # --------------------------------------------------------------------------- # 1. TEAM_BOT 매핑 정확성 # --------------------------------------------------------------------------- class TestTeamBotMapping: """TEAM_BOT 딕셔너리가 올바른 팀→봇 매핑을 갖는지 검증""" def test_dev1_team_maps_to_dev1(self, dispatch_mod): assert dispatch_mod.TEAM_BOT["dev1-team"] == "dev1" def test_dev2_team_maps_to_dev2(self, dispatch_mod): assert dispatch_mod.TEAM_BOT["dev2-team"] == "dev2" def test_dev3_team_maps_to_dev3(self, dispatch_mod): assert dispatch_mod.TEAM_BOT["dev3-team"] == "dev3" def test_team_bot_contains_all_dev_teams(self, dispatch_mod): expected_dev_teams = _load_dev_team_ids_from_org() assert len(expected_dev_teams) > 0, "organization-structure.json에서 dev 팀 로드 실패" assert set(dispatch_mod.TEAM_BOT.keys()) == expected_dev_teams # --------------------------------------------------------------------------- # 2. generate_task_id() 유니크성 # --------------------------------------------------------------------------- class TestGenerateTaskId: """generate_task_id()가 중복 없는 ID를 생성하는지 검증""" def test_first_id_is_task_1(self, dispatch_mod, tmp_path): """타이머 파일이 없을 때 첫 ID는 task-1""" task_id = dispatch_mod.generate_task_id() assert task_id == "task-1" def test_second_id_increments(self, dispatch_mod, tmp_path): """두 번째 호출은 task-2""" id1 = dispatch_mod.generate_task_id() id2 = dispatch_mod.generate_task_id() assert id1 == "task-1" assert id2 == "task-2" def test_ids_are_unique(self, dispatch_mod, tmp_path): """연속 5회 호출 시 모두 다른 ID""" ids = [dispatch_mod.generate_task_id() for _ in range(5)] assert len(set(ids)) == 5 def test_id_written_to_timer_file(self, dispatch_mod, tmp_path): """생성된 ID가 task-timers.json의 tasks 딕셔너리에 기록됨""" task_id = dispatch_mod.generate_task_id() timer_file = tmp_path / "memory" / "task-timers.json" assert timer_file.exists() data = json.loads(timer_file.read_text()) assert task_id in data["tasks"] def test_reserved_status_in_timer_file(self, dispatch_mod, tmp_path): """예약된 ID의 status가 'reserved'""" task_id = dispatch_mod.generate_task_id() timer_file = tmp_path / "memory" / "task-timers.json" data = json.loads(timer_file.read_text()) assert data["tasks"][task_id]["status"] == "reserved" def test_existing_timer_file_increments_correctly(self, dispatch_mod, tmp_path): """기존 task-timers.json에 task-3.1이 있으면 다음은 task-4""" timer_file = tmp_path / "memory" / "task-timers.json" existing = { "tasks": { "task-1.1": {"status": "completed"}, "task-2.1": {"status": "completed"}, "task-3.1": {"status": "running"}, } } timer_file.write_text(json.dumps(existing), encoding="utf-8") task_id = dispatch_mod.generate_task_id() assert task_id == "task-4" # --------------------------------------------------------------------------- # 3. build_prompt() 팀 정보 포함 여부 # --------------------------------------------------------------------------- class TestBuildPrompt: """build_prompt()가 각 팀의 올바른 정보를 포함하는지 검증""" @pytest.fixture(autouse=True) def _patch_task_file(self, dispatch_mod, tmp_path, monkeypatch): """team_prompts.build_prompt 내부의 하드코딩 경로를 tmp_path로 교체""" import prompts.team_prompts as tp original_build = tp.build_prompt def _patched_build( team_id, task_id, task_desc, level="normal", project_id=None, chain_id=None, task_type="coding" ): # task 파일 경로를 tmp_path 기준으로 교체 task_file = tmp_path / "memory" / "tasks" / f"{task_id}.md" task_file.parent.mkdir(parents=True, exist_ok=True) task_file.write_text(task_desc, encoding="utf-8") # 원본 build_prompt 대신 내부 헬퍼를 직접 호출하여 경로 우회 team = tp.TEAM_INFO.get(team_id) if not team: raise ValueError(f"알 수 없는 팀 ID: {team_id}") short_desc = task_desc[:20] timer_start = f'python3 ... start {task_id} --team {team_id} --desc "{short_desc}"' timer_end = f"python3 ... end {task_id}" report_path = f"{tmp_path}/memory/reports/{task_id}.md" if team["type"] == "direct": prompt = tp._build_direct_prompt( team, team_id, task_id, str(task_file), level, timer_start, timer_end, report_path, project_id=project_id, ) else: prompt = tp._build_glm_prompt( team, team_id, task_id, task_desc, level, timer_start, timer_end, report_path, project_id=project_id, ) # 코딩 작업에만 QC 섹션 포함 (task-202.1) if task_type == "coding": prompt += tp._build_verification_section(level) if level == "critical": prompt = ( "**[CRITICAL] 이 작업은 중요도 critical입니다. 품질 우선으로 신중하게 작업하세요.**\n\n" + prompt ) elif level == "security": prompt = "**[SECURITY] 이 작업은 보안 중요 작업입니다. 보안 최우선으로 작업하세요.**\n\n" + prompt return prompt monkeypatch.setattr(dispatch_mod, "_build_team_prompt", _patched_build) # dispatch.build_prompt가 내부적으로 _build_team_prompt를 호출하도록 # dispatch 모듈 수준에서도 교체 monkeypatch.setattr( sys.modules.get("dispatch", dispatch_mod), "_build_team_prompt", _patched_build, ) def test_dev1_team_contains_leader_name(self, dispatch_mod): prompt = dispatch_mod.build_prompt("dev1-team", "테스트 작업", "task-1.1", level="normal") assert "헤르메스" in prompt def test_dev1_team_contains_members(self, dispatch_mod): prompt = dispatch_mod.build_prompt("dev1-team", "테스트 작업", "task-1.1", level="normal") assert "불칸" in prompt assert "이리스" in prompt assert "아테나" in prompt assert "아르고스" in prompt def test_dev2_team_contains_leader_name(self, dispatch_mod): prompt = dispatch_mod.build_prompt("dev2-team", "테스트 작업", "task-1.1", level="normal") assert "오딘" in prompt def test_dev2_team_contains_members(self, dispatch_mod): prompt = dispatch_mod.build_prompt("dev2-team", "테스트 작업", "task-1.1", level="normal") assert "토르" in prompt assert "프레이야" in prompt assert "미미르" in prompt assert "헤임달" in prompt def test_dev3_team_contains_leader_name(self, dispatch_mod): prompt = dispatch_mod.build_prompt("dev3-team", "테스트 작업", "task-1.1", level="normal") assert "다그다" in prompt def test_dev3_team_contains_team_id_in_prompt(self, dispatch_mod): """dev3-team은 다그다 팀장의 direct 타입으로 팀원 목록이 포함됨""" prompt = dispatch_mod.build_prompt("dev3-team", "테스트 작업", "task-1.1", level="normal") assert "dev3-team" in prompt def test_prompt_contains_task_id(self, dispatch_mod): prompt = dispatch_mod.build_prompt("dev1-team", "테스트 작업", "task-99.1", level="normal") assert "task-99.1" in prompt # --------------------------------------------------------------------------- # 4. get_dispatch_time() # --------------------------------------------------------------------------- class TestGetDispatchTime: """get_dispatch_time()이 올바른 형식의 미래 시간을 반환하는지 검증""" def test_returns_string(self, dispatch_mod): result = dispatch_mod.get_dispatch_time(10) assert isinstance(result, str) def test_returns_parseable_datetime(self, dispatch_mod): result = dispatch_mod.get_dispatch_time(10) datetime.strptime(result, "%Y-%m-%d %H:%M:%S") def test_returns_future_time(self, dispatch_mod): from datetime import datetime as dt before = dt.now() result = dispatch_mod.get_dispatch_time(60) parsed = dt.strptime(result, "%Y-%m-%d %H:%M:%S") assert parsed > before def test_default_delay(self, dispatch_mod): """기본 delay_seconds=10일 때 미래 시간 반환""" from datetime import datetime as dt before = dt.now() result = dispatch_mod.get_dispatch_time() parsed = dt.strptime(result, "%Y-%m-%d %H:%M:%S") assert parsed > before # --------------------------------------------------------------------------- # 5. TEAM_INFO / CROSS_FUNCTIONAL 구조 검증 # --------------------------------------------------------------------------- class TestOrganizationConstants: """조직 구조 상수가 올바르게 정의되어 있는지 검증""" def test_team_info_has_all_teams(self, dispatch_mod): expected_team_count = _count_all_team_info_from_org() assert expected_team_count > 0, "organization-structure.json에서 팀 수 계산 실패" assert len(dispatch_mod.TEAM_INFO) == expected_team_count def test_each_team_info_has_required_keys(self, dispatch_mod): for team_id, info in dispatch_mod.TEAM_INFO.items(): assert "leader" in info assert "role" in info assert "members" in info def test_cross_functional_has_expected_roles(self, dispatch_mod): expected = {"qc", "redteam", "design", "devops"} assert set(dispatch_mod.CROSS_FUNCTIONAL.keys()) == expected def test_cross_functional_each_has_name_and_role(self, dispatch_mod): for key, info in dispatch_mod.CROSS_FUNCTIONAL.items(): assert "name" in info assert "role" in info # --------------------------------------------------------------------------- # 6. 잘못된 팀 ID 에러 처리 # --------------------------------------------------------------------------- class TestInvalidTeamId: """알 수 없는 팀 ID를 전달했을 때 적절한 에러가 발생하는지 검증""" def test_invalid_team_causes_sys_exit(self, dispatch_mod): """dispatch.build_prompt()는 잘못된 팀 ID에 대해 sys.exit(1) 호출""" with pytest.raises(SystemExit) as exc_info: dispatch_mod.build_prompt("nonexistent-team", "작업", "task-1.1", level="normal") assert exc_info.value.code == 1 def test_team_info_does_not_contain_invalid_key(self, dispatch_mod): """TEAM_INFO에 정의되지 않은 팀 키는 포함되지 않아야 함""" assert "nonexistent-team" not in dispatch_mod.TEAM_INFO def test_empty_team_id_causes_sys_exit(self, dispatch_mod): """빈 문자열 팀 ID도 sys.exit(1) 호출""" with pytest.raises(SystemExit) as exc_info: dispatch_mod.build_prompt("", "작업", "task-1.1") assert exc_info.value.code == 1 # --------------------------------------------------------------------------- # 7. generate_task_id() 에러 핸들링 # --------------------------------------------------------------------------- class TestGenerateTaskIdErrorHandling: """generate_task_id() 에러 핸들링 분기 테스트""" def test_corrupted_json_raises_runtime_error(self, dispatch_mod, tmp_path): """task-timers.json이 깨진 JSON일 때 RuntimeError 발생 (파일 손상 = 하드코딩 기본값 사용 금지)""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text("INVALID JSON{{{", encoding="utf-8") with pytest.raises(RuntimeError, match="손상"): dispatch_mod.generate_task_id() def test_unparseable_task_id_skipped(self, dispatch_mod, tmp_path): """숫자로 변환 불가능한 task ID가 있으면 skip (lines 110-111)""" timer_file = tmp_path / "memory" / "task-timers.json" existing = {"tasks": {"task-abc.def": {"status": "completed"}, "task-2.1": {"status": "completed"}}} timer_file.write_text(json.dumps(existing), encoding="utf-8") task_id = dispatch_mod.generate_task_id() assert task_id == "task-3" def test_timer_file_reread_failure(self, dispatch_mod, tmp_path): """task-timers.json 재읽기 실패 시 초기화 (lines 121-123)""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text('{"tasks": {"task-1.1": {"status": "done"}}}', encoding="utf-8") import builtins original_open = builtins.open call_count = [0] def mock_open_fn(*args, **kwargs): f_path = str(args[0]) if args else "" # mode 결정: keyword 우선, 없으면 두 번째 positional, 없으면 기본 'r' if len(args) > 1: mode = str(args[1]) else: mode = str(kwargs.get("mode", "r")) if "task-timers.json" in f_path and "r" in mode and "w" not in mode: call_count[0] += 1 if call_count[0] == 2: raise OSError("모킹된 읽기 실패") return original_open(*args, **kwargs) with patch("builtins.open", side_effect=mock_open_fn): task_id = dispatch_mod.generate_task_id() # 실패해도 ID가 생성되어야 함 assert task_id.startswith("task-") def test_timer_data_without_tasks_key(self, dispatch_mod, tmp_path): """timer_data에 'tasks' 키가 없을 때 자동 추가 (line 127)""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text("{}", encoding="utf-8") task_id = dispatch_mod.generate_task_id() assert task_id == "task-1" # 저장 후 tasks 키가 있어야 함 data = json.loads(timer_file.read_text()) assert "tasks" in data def test_write_failure_still_returns_id(self, dispatch_mod, tmp_path): """task-timers.json 쓰기 실패해도 ID는 반환 (lines 136-137)""" import builtins original_open = builtins.open def mock_open_write(*args, **kwargs): f_path = str(args[0]) if args else "" if len(args) > 1: mode = str(args[1]) else: mode = str(kwargs.get("mode", "r")) if "task-timers.json" in f_path and "w" in mode: raise OSError("쓰기 실패 모킹") return original_open(*args, **kwargs) with patch("builtins.open", side_effect=mock_open_write): task_id = dispatch_mod.generate_task_id() assert task_id.startswith("task-") # --------------------------------------------------------------------------- # 8. _register_followup() — 삭제됨 (task-555.1) # 사유: task-548.1에서 _register_followup이 3-Layer Defense로 대체되어 함수 삭제됨 # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # 9. dispatch() 함수 테스트 # --------------------------------------------------------------------------- class TestDispatchFunction: """dispatch() 함수 테스트 (lines 198-278)""" def _make_mock_subprocess(self, returncode=0, stdout=None, stderr=""): """subprocess mock 헬퍼""" mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"status": "ok"}) mock_result.stderr = stderr return mock_result def test_dispatch_success(self, dispatch_mod, tmp_path): """dispatch() 성공 시 status=dispatched 반환""" mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok", "id": "sched-123"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev1-team", "테스트 작업") assert result["status"] == "dispatched" assert result["team"] == "dev1-team" assert "task_id" in result def test_dispatch_failure(self, dispatch_mod, tmp_path): """dispatch() 실패 시 status=error 반환""" mock_result = self._make_mock_subprocess(1, "", "command not found") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev1-team", "테스트 작업") assert result["status"] == "error" assert "command not found" in result["message"] def test_dispatch_nonexistent_project(self, dispatch_mod, tmp_path): """존재하지 않는 project_id일 때 에러 반환""" with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = MagicMock(returncode=0, stdout="{}", stderr="") result = dispatch_mod.dispatch("dev1-team", "작업", project_id="nonexistent-proj") assert result["status"] == "error" assert "존재하지 않" in result["message"] def test_dispatch_no_bot_key(self, dispatch_mod, tmp_path): """BOT_KEYS에 봇 키가 없으면 에러""" with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {}), ): mock_sub.run.return_value = MagicMock(returncode=0, stdout="{}", stderr="") result = dispatch_mod.dispatch("dev1-team", "작업") assert result["status"] == "error" assert "할당된 봇이 없습니다" in result["message"] def test_dispatch_bot_key_none_exits(self, dispatch_mod, tmp_path): """BOT_KEYS[bot_id]가 None이면 _cleanup_task 후 error dict 반환""" with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": None, "anu": "anu-key"}), ): mock_sub.run.return_value = MagicMock(returncode=0, stdout="{}", stderr="") result = dispatch_mod.dispatch("dev1-team", "작업") assert result["status"] == "error" assert "봇 키가 설정되지 않았습니다" in result["message"] def test_dispatch_json_decode_error_in_stdout(self, dispatch_mod, tmp_path): """subprocess 성공했지만 stdout이 유효한 JSON이 아닌 경우""" mock_result = self._make_mock_subprocess(0, "not valid json output") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev1-team", "테스트") assert result["status"] == "dispatched" assert "raw" in result["cron_response"] def test_dispatch_creates_task_file(self, dispatch_mod, tmp_path): """dispatch()가 memory/tasks/.md 파일을 생성하는지""" mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev1-team", "파일 생성 테스트") task_id = result["task_id"] task_file = tmp_path / "memory" / "tasks" / f"{task_id}.md" assert task_file.exists() def test_dispatch_existing_project(self, dispatch_mod, tmp_path): """존재하는 project_id일 때 정상 dispatch""" (tmp_path / "projects" / "myproj").mkdir(parents=True, exist_ok=True) mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev1-team", "프로젝트 작업", project_id="myproj") assert result["status"] == "dispatched" def test_dispatch_returns_lead_name(self, dispatch_mod, tmp_path): """dispatch() 결과에 팀장 이름이 포함되어야 함""" mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev2-team", "작업") assert result["status"] == "dispatched" assert "오딘" in result["lead"] def test_dispatch_level_propagated(self, dispatch_mod, tmp_path): """dispatch()에 level 인자가 결과에 반영됨""" mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev1-team", "중요 작업", level="critical") assert result["level"] == "critical" # --------------------------------------------------------------------------- # 10. main() CLI 테스트 # --------------------------------------------------------------------------- class TestMainCLI: """main() CLI 함수 테스트 (lines 282-299)""" def test_main_dispatches_and_prints_json(self, dispatch_mod, monkeypatch, capsys): """main()이 dispatch() 결과를 JSON으로 출력""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "CLI 테스트"]) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert "status" in output def test_main_with_level_critical(self, dispatch_mod, monkeypatch, capsys): """main()에 --level critical 전달""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) monkeypatch.setattr( sys, "argv", ["dispatch.py", "--team", "dev2-team", "--task", "중요 작업", "--level", "critical"] ) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert output.get("level") == "critical" or "status" in output def test_main_with_session_and_project(self, dispatch_mod, monkeypatch, capsys, tmp_path): """main()에 --session, --project 전달""" # 프로젝트 디렉토리 생성 (tmp_path / "projects" / "testproj").mkdir(parents=True, exist_ok=True) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) monkeypatch.setattr( sys, "argv", [ "dispatch.py", "--team", "dev1-team", "--task", "프로젝트 작업", "--session", "sess-abc", "--project", "testproj", ], ) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert "status" in output def test_main_dev3_team(self, dispatch_mod, monkeypatch, capsys): """main()에 dev3-team 전달""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev3-team", "--task", "GLM 작업"]) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev3": "key3", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert "status" in output def test_main_marketing_team(self, dispatch_mod, monkeypatch, capsys, tmp_path): """main()에 --team marketing 전달""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "marketing", "--task", "마케팅 작업"]) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object( dispatch_mod, "BOT_KEYS", {"dev1": "key1", "dev2": "key2", "dev3": "key3", "anu": "anu-key"}, ), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert "status" in output def test_main_consulting_team(self, dispatch_mod, monkeypatch, capsys, tmp_path): """main()에 --team consulting 전달""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "consulting", "--task", "컨설팅 작업"]) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object( dispatch_mod, "BOT_KEYS", {"dev1": "key1", "dev2": "key2", "dev3": "key3", "anu": "anu-key"}, ), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert "status" in output # --------------------------------------------------------------------------- # 11. subprocess timeout 테스트 # --------------------------------------------------------------------------- class TestSubprocessTimeout: """subprocess.run() timeout 처리 테스트""" def test_dispatch_cokacdir_timeout(self, dispatch_mod, tmp_path): """cokacdir 호출에서 TimeoutExpired 발생 시 status=error 반환""" import subprocess as real_subprocess def mock_run_side_effect(*args, **kwargs): cmd = args[0] if args else kwargs.get("args", []) # cokacdir 호출에서 TimeoutExpired 발생 if isinstance(cmd, list) and cmd and cmd[0] == "cokacdir": raise real_subprocess.TimeoutExpired(cmd, 60) # 나머지(timer, log 등)는 정상 반환 mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.side_effect = mock_run_side_effect mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired result = dispatch_mod.dispatch("dev1-team", "타임아웃 테스트 작업") assert result["status"] == "error" assert "타임아웃" in result["message"] def test_dispatch_cokacdir_timeout_calls_cleanup(self, dispatch_mod, tmp_path): """cokacdir TimeoutExpired 시 _cleanup_task가 호출됨""" import subprocess as real_subprocess cleanup_called = [] def mock_cleanup(task_id): cleanup_called.append(task_id) def mock_run_side_effect(*args, **kwargs): cmd = args[0] if args else kwargs.get("args", []) if isinstance(cmd, list) and cmd and cmd[0] == "cokacdir": raise real_subprocess.TimeoutExpired(cmd, 60) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), patch.object(dispatch_mod, "_cleanup_task", side_effect=mock_cleanup), ): mock_sub.run.side_effect = mock_run_side_effect mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired dispatch_mod.dispatch("dev1-team", "cleanup 확인 테스트") assert len(cleanup_called) == 1 def test_notify_completion_send_failure_no_exit(self, tmp_path): """notify-completion.py의 Telegram 전송 실패 시 sys.exit 없이 정상 반환""" import importlib.util workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) notify_script = workspace / "scripts" / "notify-completion.py" spec = importlib.util.spec_from_file_location("notify_completion", notify_script) assert spec is not None and spec.loader is not None mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) # 체인 중간 Phase(in_chain=True, is_last=False)에서 send_telegram_notification 실패해도 sys.exit 호출 안 됨 with ( patch.object(mod, "WORKSPACE_ROOT", str(tmp_path)), patch.object( mod, "check_chain_status", return_value={"in_chain": True, "is_last": False, "chain_id": "chain-test", "next_task_id": None}, ), patch.object( mod, "dispatch_next_phase", return_value={"action": "none"}, ), patch.object(mod, "send_telegram_notification", return_value=None), patch.object(sys, "argv", ["notify-completion.py", "task-999.1"]), patch.dict(os.environ, {"COKACDIR_KEY_ANU": "test-key", "ANU_BOT_TOKEN": "test-token"}), ): # SystemExit 발생하면 안 됨 mod.main() # .done.clear 미생성 확인 (create_done_clear가 mock되어 실제 파일 미생성) assert not (events_dir / "task-999.1.done.clear").exists() def test_dispatch_timer_cmd_has_timeout_30(self, dispatch_mod, tmp_path): """로컬 python3 timer 호출은 timeout=30으로 호출됨""" import subprocess as real_subprocess run_calls = [] def mock_run_capture(*args, **kwargs): run_calls.append({"args": args[0] if args else [], "kwargs": kwargs}) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.side_effect = mock_run_capture mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired dispatch_mod.dispatch("dev1-team", "타이머 timeout 확인") # python3 호출(timer_cmd)은 timeout=30이어야 함 python3_calls = [ c for c in run_calls if isinstance(c["args"], list) and c["args"] and c["args"][0] == "python3" ] for call in python3_calls: assert call["kwargs"].get("timeout") == 30, f"python3 호출에 timeout=30 미설정: {call}" def test_dispatch_cokacdir_cmd_has_timeout_60(self, dispatch_mod, tmp_path): """외부 cokacdir 호출은 timeout=60으로 호출됨""" import subprocess as real_subprocess run_calls = [] def mock_run_capture(*args, **kwargs): run_calls.append({"args": args[0] if args else [], "kwargs": kwargs}) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.side_effect = mock_run_capture mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired dispatch_mod.dispatch("dev1-team", "cokacdir timeout 확인") # 메인 dispatch cokacdir 호출은 timeout=60이어야 함 # (wake-up ping 호출 "." 은 timeout=30으로 별도 처리되므로 제외) main_cokacdir_calls = [ c for c in run_calls if isinstance(c["args"], list) and c["args"] and c["args"][0] == "cokacdir" and len(c["args"]) >= 3 and c["args"][2] != "." ] assert len(main_cokacdir_calls) >= 1, "메인 cokacdir 호출이 없음" for call in main_cokacdir_calls: assert call["kwargs"].get("timeout") == 60, f"cokacdir 호출에 timeout=60 미설정: {call}" # --------------------------------------------------------------------------- # 12. _find_available_bot() 테스트 # --------------------------------------------------------------------------- class TestFindAvailableBot: """_find_available_bot() 가용 봇 자동 선택 테스트""" def test_all_bots_free_returns_bot_b(self, dispatch_mod, tmp_path): """모든 봇이 비어있으면 우선순위 최고인 bot-b 반환""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") result = dispatch_mod._find_available_bot() assert result == "bot-b" def test_dev1_running_returns_bot_c(self, dispatch_mod, tmp_path): """dev1-team이 running이면 bot-b 사용 중 → bot-c 반환""" timer_file = tmp_path / "memory" / "task-timers.json" data = {"tasks": {"task-1.1": {"team_id": "dev1-team", "status": "running"}}} timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._find_available_bot() assert result == "bot-c" def test_dev1_dev2_running_returns_bot_d(self, dispatch_mod, tmp_path): """dev1, dev2 모두 running → bot-d 반환""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "dev1-team", "status": "running"}, "task-2.1": {"team_id": "dev2-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._find_available_bot() assert result == "bot-d" def test_all_bots_busy_raises_error(self, dispatch_mod, tmp_path): """모든 봇이 사용 중이면 RuntimeError 발생""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "dev1-team", "status": "running"}, "task-2.1": {"team_id": "dev2-team", "status": "running"}, "task-3.1": {"team_id": "dev3-team", "status": "running"}, "task-4.1": {"team_id": "dev4-team", "status": "running"}, "task-5.1": {"team_id": "dev5-team", "status": "running"}, "task-6.1": {"team_id": "dev6-team", "status": "running"}, "task-7.1": {"team_id": "dev7-team", "status": "running"}, "task-8.1": {"team_id": "dev8-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with pytest.raises(RuntimeError, match="모든 봇이 작업 중"): dispatch_mod._find_available_bot() def test_completed_tasks_dont_block(self, dispatch_mod, tmp_path): """completed 상태의 태스크는 봇을 점유하지 않음""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "dev1-team", "status": "completed"}, "task-2.1": {"team_id": "dev2-team", "status": "completed"}, "task-3.1": {"team_id": "dev3-team", "status": "completed"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._find_available_bot() assert result == "bot-b" def test_marketing_on_bot_b_blocks_bot_b(self, dispatch_mod, tmp_path): """마케팅 태스크가 bot-b에서 running이면 bot-b 사용 불가""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "marketing", "status": "running", "bot": "bot-b"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._find_available_bot() assert result == "bot-c" def test_no_timer_file_returns_bot_b(self, dispatch_mod, tmp_path): """timer 파일이 없으면 모든 봇 가용 → bot-b""" timer_file = tmp_path / "memory" / "task-timers.json" if timer_file.exists(): timer_file.unlink() result = dispatch_mod._find_available_bot() assert result == "bot-b" def test_corrupted_json_returns_bot_b(self, dispatch_mod, tmp_path): """timer 파일이 깨진 JSON이면 모든 봇 가용으로 처리""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text("INVALID JSON{{{", encoding="utf-8") result = dispatch_mod._find_available_bot() assert result == "bot-b" # --------------------------------------------------------------------------- # 13. _patch_timer_metadata() 테스트 # --------------------------------------------------------------------------- class TestPatchTimerMetadata: """_patch_timer_metadata() 메타데이터 패치 테스트""" def test_patches_existing_task(self, dispatch_mod, tmp_path): """기존 태스크에 role/bot 메타데이터 추가""" timer_file = tmp_path / "memory" / "task-timers.json" data = {"tasks": {"task-1.1": {"team_id": "marketing", "status": "running"}}} timer_file.write_text(json.dumps(data), encoding="utf-8") dispatch_mod._patch_timer_metadata("task-1.1", role="marketing", bot="bot-b") result = json.loads(timer_file.read_text()) assert result["tasks"]["task-1.1"]["role"] == "marketing" assert result["tasks"]["task-1.1"]["bot"] == "bot-b" def test_nonexistent_task_no_error(self, dispatch_mod, tmp_path): """존재하지 않는 태스크에 패치 시 에러 없음""" timer_file = tmp_path / "memory" / "task-timers.json" data = {"tasks": {}} timer_file.write_text(json.dumps(data), encoding="utf-8") dispatch_mod._patch_timer_metadata("task-999.1", role="marketing", bot="bot-b") result = json.loads(timer_file.read_text()) assert "task-999.1" not in result["tasks"] def test_no_timer_file_no_error(self, dispatch_mod, tmp_path): """timer 파일이 없어도 에러 없음""" timer_file = tmp_path / "memory" / "task-timers.json" if timer_file.exists(): timer_file.unlink() # 에러 없이 실행되어야 함 dispatch_mod._patch_timer_metadata("task-1.1", role="dev1", bot="bot-b") def test_preserves_existing_fields(self, dispatch_mod, tmp_path): """기존 필드는 유지하면서 새 필드 추가""" timer_file = tmp_path / "memory" / "task-timers.json" data = {"tasks": {"task-1.1": {"team_id": "dev1-team", "status": "running", "description": "test"}}} timer_file.write_text(json.dumps(data), encoding="utf-8") dispatch_mod._patch_timer_metadata("task-1.1", role="dev1", bot="bot-b") result = json.loads(timer_file.read_text()) assert result["tasks"]["task-1.1"]["team_id"] == "dev1-team" assert result["tasks"]["task-1.1"]["description"] == "test" assert result["tasks"]["task-1.1"]["role"] == "dev1" # --------------------------------------------------------------------------- # 14. 마케팅/컨설팅 dispatch 테스트 # --------------------------------------------------------------------------- class TestMarketingConsultingDispatch: """마케팅/컨설팅 팀 dispatch 테스트""" def _make_mock_subprocess(self, returncode=0, stdout=None, stderr=""): mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"status": "ok"}) mock_result.stderr = stderr return mock_result def test_marketing_dispatch_success(self, dispatch_mod, tmp_path): """marketing 팀 dispatch 성공""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object( dispatch_mod, "BOT_KEYS", {"dev1": "key1", "dev2": "key2", "dev3": "key3", "anu": "anu-key"}, ), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("marketing", "마케팅 콘텐츠 작성") assert result["status"] == "dispatched" assert result["team"] == "marketing" def test_consulting_dispatch_success(self, dispatch_mod, tmp_path): """consulting 팀 dispatch 성공""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object( dispatch_mod, "BOT_KEYS", {"dev1": "key1", "dev2": "key2", "dev3": "key3", "anu": "anu-key"}, ), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("consulting", "보험 약관 비교") assert result["status"] == "dispatched" assert result["team"] == "consulting" def test_marketing_all_bots_busy_returns_error(self, dispatch_mod, tmp_path): """모든 봇이 사용 중일 때 marketing dispatch → error""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "dev1-team", "status": "running"}, "task-2.1": {"team_id": "dev2-team", "status": "running"}, "task-3.1": {"team_id": "dev3-team", "status": "running"}, "task-4.1": {"team_id": "dev4-team", "status": "running"}, "task-5.1": {"team_id": "dev5-team", "status": "running"}, "task-6.1": {"team_id": "dev6-team", "status": "running"}, "task-7.1": {"team_id": "dev7-team", "status": "running"}, "task-8.1": {"team_id": "dev8-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object( dispatch_mod, "BOT_KEYS", { "dev1": "key1", "dev2": "key2", "dev3": "key3", "dev4": "key4", "dev5": "key5", "dev6": "key6", "dev7": "key7", "dev8": "key8", "anu": "anu-key", }, ), ): mock_sub.run.return_value = MagicMock(returncode=0, stdout="{}", stderr="") result = dispatch_mod.dispatch("marketing", "마케팅 작업") assert result["status"] == "error" assert "모든 봇이 작업 중" in result["message"] def test_marketing_bot_key_none_returns_error(self, dispatch_mod, tmp_path): """선택된 봇의 키가 None이면 error 반환 (sys.exit 아님)""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") with patch.object( dispatch_mod, "BOT_KEYS", {"dev1": None, "dev2": "key2", "dev3": "key3", "anu": "anu-key"}, ): result = dispatch_mod.dispatch("marketing", "봇 키 테스트") assert result["status"] == "error" assert "봇 키가 설정되지 않았습니다" in result["message"] def test_marketing_metadata_recorded(self, dispatch_mod, tmp_path): """마케팅 dispatch 시 role/bot 메타데이터 기록""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object( dispatch_mod, "BOT_KEYS", {"dev1": "key1", "dev2": "key2", "dev3": "key3", "anu": "anu-key"}, ), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("marketing", "메타데이터 테스트") # task-timers.json에 role/bot이 기록되었는지 확인 data = json.loads(timer_file.read_text()) task_id = result["task_id"] # timer_task_id는 task_id에서 .이 없으면 .1을 붙이므로 실제 키를 찾아야 함 timer_key = task_id if "." in task_id else f"{task_id}.1" if timer_key in data["tasks"]: assert data["tasks"][timer_key].get("role") == "marketing" assert data["tasks"][timer_key].get("bot") in ("bot-b", "bot-c", "bot-d") # --------------------------------------------------------------------------- # 15. task_desc 첫 줄 task-id 교체 테스트 (TDD RED) # --------------------------------------------------------------------------- class TestTaskIdReplacement: """task_desc 첫 줄의 플레이스홀더 task-id가 실제 task_id로 교체되어 저장되어야 함. 현재 dispatch.py는 task_desc를 그대로 저장하므로 이 테스트는 실패(RED). """ def _make_mock_subprocess(self, returncode=0, stdout=None, stderr=""): mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"status": "ok"}) mock_result.stderr = stderr return mock_result def test_task_desc_first_line_task_id_replaced(self, dispatch_mod, tmp_path): """task_desc 첫 줄에 '# task-999.1: 제목' 패턴이 있으면 저장된 파일의 첫 줄이 실제 task_id로 교체되어야 함. 예: task_desc = "# task-999.1: 테스트 작업\n\n내용" 생성된 task_id가 task-1.1이면 저장 파일 첫 줄 = "# task-1.1: 테스트 작업" """ task_desc = "# task-999.1: 테스트 작업\n\n상세 내용이 여기 있습니다." mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev2-team", task_desc) assert result["status"] == "dispatched" actual_task_id = result["task_id"] task_file = tmp_path / "memory" / "tasks" / f"{actual_task_id}.md" assert task_file.exists() saved_content = task_file.read_text(encoding="utf-8") first_line = saved_content.splitlines()[0] # 실제 task_id로 교체되어야 함 (999.1이 아닌 actual_task_id) assert ( first_line == f"# {actual_task_id}: 테스트 작업" ), f"첫 줄이 교체되지 않음. 기대: '# {actual_task_id}: 테스트 작업', 실제: '{first_line}'" def test_task_desc_without_task_id_unchanged(self, dispatch_mod, tmp_path): """task_desc 첫 줄에 task-id 패턴이 없으면 내용이 그대로 저장되어야 함.""" task_desc = "# 일반 제목\n\n내용이 여기 있습니다." mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev2-team", task_desc) assert result["status"] == "dispatched" actual_task_id = result["task_id"] task_file = tmp_path / "memory" / "tasks" / f"{actual_task_id}.md" assert task_file.exists() saved_content = task_file.read_text(encoding="utf-8") first_line = saved_content.splitlines()[0] # 패턴 없으면 변경 없어야 함 assert first_line == "# 일반 제목", f"첫 줄이 의도치 않게 변경됨. 기대: '# 일반 제목', 실제: '{first_line}'" def test_task_desc_with_explicit_task_id_flag(self, dispatch_mod, tmp_path): """--task-id로 task_id를 직접 지정한 경우에도 task_desc 첫 줄의 플레이스홀더가 지정된 task_id로 교체되어야 함. 예: task_id="task-42.1", task_desc="# task-999.1: 명시 지정 테스트\n\n내용" → 저장 파일 첫 줄 = "# task-42.1: 명시 지정 테스트" """ task_desc = "# task-999.1: 명시 지정 테스트\n\n상세 내용." explicit_task_id = "task-42.1" mock_result = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) # task-timers.json에 미리 placeholder 등록 (generate_task_id 우회 시 필요) timer_file = tmp_path / "memory" / "task-timers.json" timer_data = {"tasks": {explicit_task_id: {"status": "reserved", "reserved_at": "2026-01-01T00:00:00"}}} timer_file.write_text(json.dumps(timer_data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch("dev2-team", task_desc, task_id=explicit_task_id) assert result["status"] == "dispatched" assert result["task_id"] == explicit_task_id task_file = tmp_path / "memory" / "tasks" / f"{explicit_task_id}.md" assert task_file.exists() saved_content = task_file.read_text(encoding="utf-8") first_line = saved_content.splitlines()[0] # 명시 지정된 task_id로 교체되어야 함 assert ( first_line == f"# {explicit_task_id}: 명시 지정 테스트" ), f"첫 줄이 교체되지 않음. 기대: '# {explicit_task_id}: 명시 지정 테스트', 실제: '{first_line}'" # --------------------------------------------------------------------------- # 16. dispatch 실패 시 task-timers.json 정리 테스트 (TDD RED) # --------------------------------------------------------------------------- class TestDispatchFailureCleanup: """dispatch() 실패 시 task-timers.json에 잔여 task_id가 없어야 함. 현재 dispatch.py에서 project_id 디렉토리 없음(line 412)은 _cleanup_task를 호출하지 않고 바로 return하며, dev팀 봇 키 None(line 475)은 sys.exit(1)을 호출하므로 두 경우 모두 task-timers.json에 orphan 항목이 남는 버그가 있음(RED). """ def _setup_timer_file(self, tmp_path, extra_tasks=None): """테스트용 빈 task-timers.json 초기화""" timer_file = tmp_path / "memory" / "task-timers.json" data = {"tasks": extra_tasks or {}} timer_file.write_text(json.dumps(data), encoding="utf-8") return timer_file def _get_timer_tasks(self, tmp_path): """현재 task-timers.json의 tasks 딕셔너리 반환""" timer_file = tmp_path / "memory" / "task-timers.json" if not timer_file.exists(): return {} return json.loads(timer_file.read_text(encoding="utf-8")).get("tasks", {}) def test_cleanup_on_project_dir_not_exists(self, dispatch_mod, tmp_path): """project_id를 지정했는데 해당 디렉토리가 없으면 error를 반환하고 task-timers.json에 해당 task_id가 남아있지 않아야 함. 현재 dispatch.py line 412에서 _cleanup_task 없이 return → orphan 발생(버그). """ self._setup_timer_file(tmp_path) # subprocess는 task-timer start 호출을 위해 mock 필요 mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" mock_result.stderr = "" with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result result = dispatch_mod.dispatch( "dev1-team", "프로젝트 없는 작업", project_id="nonexistent-project-xyz", ) # 반환값은 error여야 함 assert result["status"] == "error" assert "존재하지 않" in result["message"] # task-timers.json에 이 dispatch에서 생성된 task_id가 남아있으면 안 됨 tasks = self._get_timer_tasks(tmp_path) # reserved 또는 running 상태인 항목이 없어야 함 orphan_ids = [tid for tid, entry in tasks.items() if entry.get("status") in ("reserved", "running")] assert len(orphan_ids) == 0, f"실패 후 task-timers.json에 orphan 항목이 남아있음: {orphan_ids}" def test_cleanup_on_bot_key_missing(self, dispatch_mod, tmp_path): """dev팀 봇 키가 None일 때 error를 반환하고 task-timers.json에 해당 task_id가 남아있지 않아야 함. 현재 dispatch.py line 475에서 sys.exit(1) 호출 → _cleanup_task 미호출(버그). 이 테스트는 sys.exit 대신 error dict를 반환하는 동작을 기대함. """ self._setup_timer_file(tmp_path) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" mock_result.stderr = "" # dev2 봇 키를 None으로 설정 with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": None, "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result # 현재 코드는 sys.exit(1)을 호출하므로 이 테스트는 SystemExit 대신 # error dict 반환을 기대 → RED (sys.exit를 error return으로 변경 필요) result = dispatch_mod.dispatch("dev2-team", "봇 키 없는 작업") # 반환값은 error여야 함 (sys.exit가 아닌 dict 반환) assert isinstance(result, dict), "봇 키 None 시 sys.exit 대신 error dict를 반환해야 함" assert result["status"] == "error" # task-timers.json에 이 dispatch에서 생성된 task_id가 남아있으면 안 됨 tasks = self._get_timer_tasks(tmp_path) orphan_ids = [tid for tid, entry in tasks.items() if entry.get("status") in ("reserved", "running")] assert len(orphan_ids) == 0, f"실패 후 task-timers.json에 orphan 항목이 남아있음: {orphan_ids}" # --------------------------------------------------------------------------- # 17. TestDispatchPhasedChaining: --phases 옵션 체이닝 테스트 # --------------------------------------------------------------------------- class TestDispatchPhasedChaining: """dispatch() + phases 파라미터로 chain_manager.py create를 호출하는 테스트""" def _make_dispatch_mock(self, returncode: int = 0, stdout: str = ""): """subprocess.run mock 결과를 생성한다.""" mock = MagicMock() mock.returncode = returncode mock.stdout = stdout mock.stderr = "" return mock def test_dispatch_with_phases_creates_chain(self, dispatch_mod, tmp_path): """--phases 5로 dispatch 시 chain_manager.py create가 subprocess로 호출된다.""" import subprocess as real_subprocess run_calls = [] def mock_run(*args, **kwargs): run_calls.append({"args": args[0] if args else [], "kwargs": kwargs}) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.side_effect = mock_run mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired result = dispatch_mod.dispatch( "dev1-team", "# task-566.1: 체이닝 테스트", task_id="task-566.1", phases=5, ) # dispatch 자체는 성공해야 함 assert result["status"] == "dispatched" # chain_manager.py create 호출 확인 all_cmds = [str(c["args"]) for c in run_calls] chain_create_calls = [ c for c in run_calls if "chain_manager.py" in str(c["args"]) and "create" in str(c["args"]) ] assert len(chain_create_calls) >= 1, f"chain_manager.py create 호출이 없음. 실제 호출: {all_cmds}" def test_dispatch_phases_generates_correct_chain_tasks(self, dispatch_mod, tmp_path): """phases=3이면 task-566.1, task-566.2, task-566.3 형식의 tasks JSON이 생성된다.""" import subprocess as real_subprocess captured_chain_args: list = [] def mock_run(*args, **kwargs): cmd = args[0] if args else [] if "chain_manager.py" in str(cmd) and "create" in str(cmd): captured_chain_args.append(cmd) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.side_effect = mock_run mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired dispatch_mod.dispatch( "dev1-team", "# task-566.1: 체이닝 테스트", task_id="task-566.1", phases=3, ) assert len(captured_chain_args) >= 1, "chain_manager.py create 호출이 없음" # --tasks 인자에서 JSON 파싱 cmd = captured_chain_args[0] tasks_idx = cmd.index("--tasks") + 1 tasks_json = cmd[tasks_idx] tasks = json.loads(tasks_json) assert len(tasks) == 3 assert tasks[0]["task_id"] == "task-566.1" assert tasks[1]["task_id"] == "task-566.2" assert tasks[2]["task_id"] == "task-566.3" assert tasks[0]["order"] == 1 assert tasks[1]["order"] == 2 assert tasks[2]["order"] == 3 def test_dispatch_phases_extracts_base_from_task_id(self, dispatch_mod, tmp_path): """task_id에서 base 번호를 올바르게 추출하여 chain_id를 생성한다.""" import subprocess as real_subprocess captured_chain_args: list = [] def mock_run(*args, **kwargs): cmd = args[0] if args else [] if "chain_manager.py" in str(cmd) and "create" in str(cmd): captured_chain_args.append(cmd) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.side_effect = mock_run mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired dispatch_mod.dispatch( "dev1-team", "# task-566.1: 체이닝 테스트", task_id="task-566.1", phases=2, ) assert len(captured_chain_args) >= 1, "chain_manager.py create 호출이 없음" cmd = captured_chain_args[0] # --chain-id 인자 확인 chain_id_idx = cmd.index("--chain-id") + 1 chain_id = cmd[chain_id_idx] # chain_id는 "scoped-566" 형식이어야 함 assert chain_id == "scoped-566", f"chain_id가 예상값과 다름: {chain_id}" def test_dispatch_chain_failure_does_not_abort_dispatch(self, dispatch_mod, tmp_path): """chain_manager.py create 실패 시에도 dispatch 자체는 성공해야 한다.""" import subprocess as real_subprocess call_count = [0] def mock_run(*args, **kwargs): cmd = args[0] if args else [] call_count[0] += 1 mock_result = MagicMock() if "chain_manager.py" in str(cmd) and "create" in str(cmd): # chain_manager.py create는 실패 mock_result.returncode = 1 mock_result.stdout = "" mock_result.stderr = "chain create error" else: mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.side_effect = mock_run mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired result = dispatch_mod.dispatch( "dev1-team", "# task-566.1: 체이닝 테스트", task_id="task-566.1", phases=5, ) # chain 실패에도 불구하고 dispatch 자체는 성공 assert result["status"] == "dispatched" def test_dispatch_result_includes_chain_id_when_phases_set(self, dispatch_mod, tmp_path): """phases 지정 시 dispatch 결과에 chain_id가 포함된다.""" import subprocess as real_subprocess def mock_run(*args, **kwargs): mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" return mock_result with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.side_effect = mock_run mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired result = dispatch_mod.dispatch( "dev1-team", "# task-566.1: 체이닝 테스트", task_id="task-566.1", phases=3, ) assert result["status"] == "dispatched" assert "chain_id" in result, "dispatch 결과에 chain_id가 없음" assert result["chain_id"] == "scoped-566" # --------------------------------------------------------------------------- # 18. P1 — dispatch 병렬실행 강화: warning → block 전환 + --force 플래그 (TDD) # --------------------------------------------------------------------------- class TestDispatchParallelBlock: """동일 팀에 running 태스크가 있을 때 dispatch 거부(force=False) / 허용(force=True) 테스트. P1 요구사항 (task-792.1): - force=False(기본값): 동일 팀 running 태스크가 있으면 status="error" 반환 + task_id 포함 - force=True: 기존처럼 경고만 로깅하고 dispatch 진행 - running 태스크 없으면: force 무관하게 정상 통과 - cleanup 보장: 거부 시에도 _cleanup_task() 호출 필수 """ def _make_mock_subprocess(self, returncode=0, stdout=None, stderr=""): mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"status": "ok"}) mock_result.stderr = stderr return mock_result def _setup_running_task(self, tmp_path: Path, team_id: str = "dev1-team", running_task_id: str = "task-100.1"): """task-timers.json에 지정 팀의 running 태스크를 등록한다.""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { running_task_id: { "team_id": team_id, "status": "running", "description": "기존 진행 중 작업", } } } timer_file.write_text(json.dumps(data), encoding="utf-8") return running_task_id def test_running_task_same_team_force_false_returns_error(self, dispatch_mod, tmp_path): """동일 팀에 running 태스크가 있고 force=False이면 status='error' 반환해야 함.""" running_id = self._setup_running_task(tmp_path, team_id="dev1-team") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev1-team", "새 작업", force=False) assert result["status"] == "error", f"force=False 시 error 반환 기대, 실제: {result}" def test_running_task_same_team_error_message_contains_running_task_id(self, dispatch_mod, tmp_path): """에러 메시지에 현재 running 태스크 ID가 포함되어야 함.""" running_id = self._setup_running_task(tmp_path, team_id="dev1-team", running_task_id="task-100.1") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev1-team", "새 작업", force=False) assert result["status"] == "error" assert ( "task-100.1" in result["message"] ), f"에러 메시지에 running task ID 'task-100.1' 미포함. 실제 메시지: {result['message']}" def test_running_task_same_team_force_true_allows_dispatch(self, dispatch_mod, tmp_path): """동일 팀 running 태스크가 있어도 force=True이면 dispatch가 허용되어야 함.""" self._setup_running_task(tmp_path, team_id="dev1-team") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev1-team", "강제 작업", force=True) assert result["status"] == "dispatched", f"force=True 시 dispatched 기대, 실제: {result}" def test_no_running_task_dispatches_normally(self, dispatch_mod, tmp_path): """running 태스크가 없으면 force=False라도 정상 dispatch 되어야 함.""" # 빈 task-timers.json (running 없음) timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev1-team", "일반 작업", force=False) assert result["status"] == "dispatched" def test_running_task_different_team_does_not_block(self, dispatch_mod, tmp_path): """다른 팀에 running 태스크가 있어도 현재 팀은 차단하지 않아야 함.""" # dev2-team에 running 태스크 self._setup_running_task(tmp_path, team_id="dev2-team", running_task_id="task-200.1") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) # dev1-team에 dispatch → 차단하면 안 됨 result = dispatch_mod.dispatch("dev1-team", "다른 팀 있을 때 작업", force=False) assert result["status"] == "dispatched", f"다른 팀 running 태스크가 차단하면 안 됨. 실제: {result}" def test_block_calls_cleanup_task(self, dispatch_mod, tmp_path): """force=False로 거부 시 _cleanup_task()가 반드시 호출되어야 함.""" self._setup_running_task(tmp_path, team_id="dev1-team") cleanup_called = [] def mock_cleanup(task_id): cleanup_called.append(task_id) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), patch.object(dispatch_mod, "_cleanup_task", side_effect=mock_cleanup), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev1-team", "새 작업", force=False) assert result["status"] == "error" assert len(cleanup_called) == 1, f"_cleanup_task가 정확히 1회 호출되어야 함, 실제: {cleanup_called}" def test_force_true_only_logs_warning_not_error(self, dispatch_mod, tmp_path): """force=True일 때 경고만 로깅하고 error를 반환하지 않아야 함.""" self._setup_running_task(tmp_path, team_id="dev2-team") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev2-team", "force 허용 작업", force=True) # error가 아닌 dispatched여야 함 assert result["status"] != "error", f"force=True 시 에러 반환하면 안 됨. 실제: {result}" assert result["status"] == "dispatched" def test_force_default_is_false(self, dispatch_mod, tmp_path): """force 파라미터 기본값은 False이어야 함 (명시하지 않으면 거부 동작).""" self._setup_running_task(tmp_path, team_id="dev1-team") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) # force 인자를 명시하지 않음 → 기본값 False → 거부 result = dispatch_mod.dispatch("dev1-team", "기본값 테스트") assert result["status"] == "error", "force 기본값이 False여야 하며, running 태스크 있으면 거부해야 함" class TestDispatchForceCLI: """--force CLI 플래그 테스트.""" def test_main_with_force_flag_passes_force_true(self, dispatch_mod, monkeypatch, tmp_path): """--force 플래그가 있으면 dispatch()에 force=True가 전달되어야 함.""" # running 태스크 등록 timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text( json.dumps( { "tasks": { "task-100.1": { "team_id": "dev1-team", "status": "running", "description": "기존 작업", } } } ), encoding="utf-8", ) monkeypatch.setattr( sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "force CLI 테스트", "--force"] ) captured_force = [] original_dispatch = dispatch_mod.dispatch def mock_dispatch(*args, **kwargs): captured_force.append(kwargs.get("force", False)) # force=True이면 정상 처리 흉내 return { "status": "dispatched", "task_id": "task-1.1", "team": "dev1-team", "lead": "헤르메스", "level": "normal", "description": "force CLI 테스트", "message": "ok", "cron_response": {}, } monkeypatch.setattr(dispatch_mod, "dispatch", mock_dispatch) import io from contextlib import redirect_stdout f = io.StringIO() with redirect_stdout(f): dispatch_mod.main() assert len(captured_force) == 1, "dispatch()가 정확히 1번 호출되어야 함" assert captured_force[0] is True, f"--force 플래그 시 force=True 전달 기대, 실제: {captured_force[0]}" def test_main_without_force_flag_passes_force_false(self, dispatch_mod, monkeypatch, tmp_path): """--force 플래그가 없으면 dispatch()에 force=False가 전달되어야 함.""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "일반 CLI 테스트"]) captured_force = [] def mock_dispatch(*args, **kwargs): captured_force.append(kwargs.get("force", False)) return { "status": "dispatched", "task_id": "task-1.1", "team": "dev1-team", "lead": "헤르메스", "level": "normal", "description": "일반 CLI 테스트", "message": "ok", "cron_response": {}, } monkeypatch.setattr(dispatch_mod, "dispatch", mock_dispatch) import io from contextlib import redirect_stdout f = io.StringIO() with redirect_stdout(f): dispatch_mod.main() assert len(captured_force) == 1 assert captured_force[0] is False, f"--force 없을 때 force=False 전달 기대, 실제: {captured_force[0]}" # --------------------------------------------------------------------------- # 테스트: _warn_research_impl_mix (리서치-구현 혼합 감지 가드) # --------------------------------------------------------------------------- class TestWarnResearchImplMix: """_warn_research_impl_mix 함수의 WARNING 로그 출력 검증""" def test_mixed_research_impl_emits_warning(self, dispatch_mod): """리서치+구현 키워드가 모두 포함된 경우 WARNING 출력""" mixed_desc = "API 문서를 조사하고 Publisher 파이프라인을 구현하세요" with patch.object(dispatch_mod.logger, "warning") as mock_warn: dispatch_mod._warn_research_impl_mix(mixed_desc, "coding") mock_warn.assert_called_once() call_msg = mock_warn.call_args[0][0] assert "[research-impl-mix]" in call_msg def test_impl_only_no_warning(self, dispatch_mod): """구현 키워드만 있을 때는 WARNING 미출력""" impl_desc = "Publisher 파이프라인을 구현하고 테스트 작성하세요" with patch.object(dispatch_mod.logger, "warning") as mock_warn: dispatch_mod._warn_research_impl_mix(impl_desc, "coding") mock_warn.assert_not_called() def test_research_only_no_warning(self, dispatch_mod): """리서치 키워드만 있을 때는 WARNING 미출력""" research_desc = "API 문서를 조사하고 인증 방식을 파악하세요" with patch.object(dispatch_mod.logger, "warning") as mock_warn: dispatch_mod._warn_research_impl_mix(research_desc, "coding") mock_warn.assert_not_called() def test_mixed_but_research_type_no_warning(self, dispatch_mod): """리서치+구현 혼합이지만 task_type이 research이면 WARNING 미출력""" mixed_desc = "API 문서를 조사하고 Publisher 파이프라인을 구현하세요" with patch.object(dispatch_mod.logger, "warning") as mock_warn: dispatch_mod._warn_research_impl_mix(mixed_desc, "research") mock_warn.assert_not_called() # --------------------------------------------------------------------------- # 19. _warn_large_task_desc + _warn_research_impl_mix 세션 경량화 메시지 (TDD) # --------------------------------------------------------------------------- def test_warn_large_task_desc(caplog): """3000자 이상 지시서에 대해 WARNING 로그 발생 확인""" import logging import sys from pathlib import Path workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) import dispatch as dispatch_mod large_desc = "A" * 3001 with caplog.at_level(logging.WARNING): dispatch_mod._warn_large_task_desc(large_desc) warning_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING] assert any( "[large-task-desc]" in msg for msg in warning_messages ), f"WARNING 로그에 '[large-task-desc]' 미포함. 기록된 로그: {warning_messages}" def test_warn_large_task_desc_under_3000(caplog): """3000자 미만 지시서에 대해 WARNING 로그 없음 확인""" import logging import sys from pathlib import Path workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) import dispatch as dispatch_mod small_desc = "A" * 2999 with caplog.at_level(logging.WARNING): dispatch_mod._warn_large_task_desc(small_desc) warning_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING] assert not any( "[large-task-desc]" in msg for msg in warning_messages ), f"3000자 미만인데 WARNING 로그 발생. 기록된 로그: {warning_messages}" def test_warn_research_impl_mix_session_msg(caplog): """리서치+구현 혼합 경고에 세션 경량화 참조 메시지 포함 확인""" import logging import sys from pathlib import Path workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) import dispatch as dispatch_mod mixed_desc = "API 문서를 조사하고 Publisher 파이프라인을 구현하세요" with caplog.at_level(logging.WARNING): dispatch_mod._warn_research_impl_mix(mixed_desc, "coding") warning_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING] assert any( "/compact" in msg for msg in warning_messages ), f"WARNING 로그에 '/compact' 미포함. 기록된 로그: {warning_messages}" # --------------------------------------------------------------------------- # 20. TestCounterBasedTaskId: 카운터 파일 기반 채번 테스트 (v2 방어 로직) # --------------------------------------------------------------------------- class TestCounterBasedTaskId: """카운터 파일 기반 generate_task_id() 테스트 (v2 채번 방어 로직)""" def test_counter_file_determines_next_id(self, dispatch_mod, tmp_path): """카운터 파일이 있으면 그 값으로 ID 생성""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("100") timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}})) task_id = dispatch_mod.generate_task_id() assert task_id == "task-100" def test_counter_increments_after_generation(self, dispatch_mod, tmp_path): """ID 생성 후 카운터 파일이 +1 증가""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("50") timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}})) dispatch_mod.generate_task_id() assert counter_file.read_text().strip() == "51" def test_consecutive_ids_from_counter(self, dispatch_mod, tmp_path): """연속 호출 시 카운터 기반으로 연속 ID 생성""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("10") timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}})) ids = [dispatch_mod.generate_task_id() for _ in range(3)] assert ids == ["task-10", "task-11", "task-12"] def test_corrupted_counter_falls_back_to_timers(self, dispatch_mod, tmp_path): """카운터 파일이 손상되면 task-timers.json에서 복구""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("not_a_number") timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {"task-5.1": {"status": "completed"}}})) task_id = dispatch_mod.generate_task_id() assert task_id == "task-6" def test_missing_counter_falls_back_to_timers(self, dispatch_mod, tmp_path): """카운터 파일 없으면 task-timers.json에서 계산""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {"task-3.1": {"status": "completed"}}})) task_id = dispatch_mod.generate_task_id() assert task_id == "task-4" def test_counter_outlier_1000_over_timers_corrected(self, dispatch_mod, tmp_path): """카운터(99999)가 timers 최대(4) 대비 1000 이상 큰 경우 → timers 기준으로 보정""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("99999") timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {"task-3": {"status": "completed"}}})) task_id = dispatch_mod.generate_task_id() assert task_id == "task-4" # --------------------------------------------------------------------------- # 21. TestOutlierFiltering: 이상치 ID 필터링 테스트 # --------------------------------------------------------------------------- class TestOutlierFiltering: """이상치 ID 필터링 테스트 — 비정상 큰 ID가 채번을 오염시키지 않는지 검증""" def test_large_gap_ids_filtered(self, dispatch_mod, tmp_path): """정상 ID + 비정상적으로 큰 ID가 있을 때 정상 ID 기반으로 채번""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text( json.dumps( { "tasks": { "task-1.1": {"status": "completed"}, "task-2.1": {"status": "completed"}, "task-3.1": {"status": "completed"}, "task-9991.1": {"status": "completed"}, # 이상치 "task-9992.1": {"status": "completed"}, # 이상치 } } ) ) # 카운터 파일 없음 → fallback으로 이상치 필터링 적용 task_id = dispatch_mod.generate_task_id() assert task_id == "task-4" # 9991/9992 무시, 3+1=4 def test_no_outliers_normal_behavior(self, dispatch_mod, tmp_path): """이상치 없으면 기존처럼 max+1""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text( json.dumps( { "tasks": { "task-1.1": {"status": "completed"}, "task-2.1": {"status": "completed"}, "task-3.1": {"status": "running"}, } } ) ) task_id = dispatch_mod.generate_task_id() assert task_id == "task-4" def test_counter_takes_precedence_over_outliers(self, dispatch_mod, tmp_path): """카운터 파일이 있으면 이상치와 무관하게 카운터 사용""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("100") timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text( json.dumps( { "tasks": { "task-1.1": {"status": "completed"}, "task-9999.1": {"status": "completed"}, # 이상치 } } ) ) task_id = dispatch_mod.generate_task_id() assert task_id == "task-100" # 카운터 우선 def test_gap_threshold_is_1000(self, dispatch_mod, tmp_path): """갭 999는 정상, 갭 1000 이상은 이상치로 판정""" timer_file = tmp_path / "memory" / "task-timers.json" # 갭 999: 1 → 1000 (999 차이) → 정상 timer_file.write_text( json.dumps( { "tasks": { "task-1.1": {"status": "completed"}, "task-1000.1": {"status": "completed"}, # 갭 999 → 정상 } } ) ) task_id = dispatch_mod.generate_task_id() assert task_id == "task-1001" # 1000은 정상으로 인정 def test_gap_exactly_1000_is_outlier(self, dispatch_mod, tmp_path): """갭이 정확히 1000이면 이상치""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text( json.dumps( { "tasks": { "task-1.1": {"status": "completed"}, "task-1001.1": {"status": "completed"}, # 갭 1000 → 이상치 } } ) ) task_id = dispatch_mod.generate_task_id() assert task_id == "task-2" # 1001 무시 # --------------------------------------------------------------------------- # 22. TestCounterFileEdgeCases: 카운터 파일 엣지 케이스 # --------------------------------------------------------------------------- class TestCounterFileEdgeCases: """카운터 파일 엣지 케이스 테스트""" def test_counter_file_created_on_first_run(self, dispatch_mod, tmp_path): """첫 실행 시 카운터 파일이 생성됨""" counter_file = tmp_path / "memory" / ".task-counter" assert not counter_file.exists() dispatch_mod.generate_task_id() assert counter_file.exists() def test_empty_counter_file_falls_back(self, dispatch_mod, tmp_path): """빈 카운터 파일은 fallback""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("") timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {"task-7.1": {"status": "completed"}}})) task_id = dispatch_mod.generate_task_id() assert task_id == "task-8" def test_negative_counter_falls_back(self, dispatch_mod, tmp_path): """음수 카운터도 그대로 사용 (방어는 하되 동작은 유지)""" # 이 케이스는 구현에 따라 fallback 또는 사용 가능 counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("-5") timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}})) # 음수 ID는 비정상이므로 fallback이 바람직 # 구현에 따라 assert 조정 가능 # --------------------------------------------------------------------------- # 23. TestSyncCounterPhaseAware: _sync_counter_if_needed() Phase 인식 테스트 # --------------------------------------------------------------------------- class TestSyncCounterPhaseAware: """_sync_counter_if_needed()가 Phase 접미사를 올바르게 무시하는지 검증""" def test_phase_suffix_ignored(self, dispatch_mod, tmp_path): """task-1845_2.2 → 기본 번호 1845만 추출하여 카운터 sync""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("100") dispatch_mod._sync_counter_if_needed("task-1845_2.2") assert counter_file.read_text().strip() == "1846" def test_parallel_suffix_ignored(self, dispatch_mod, tmp_path): """task-500_a → 기본 번호 500만 추출""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("100") dispatch_mod._sync_counter_if_needed("task-500_a") assert counter_file.read_text().strip() == "501" def test_retry_suffix_ignored(self, dispatch_mod, tmp_path): """task-300+1 → 기본 번호 300만 추출""" counter_file = tmp_path / "memory" / ".task-counter" counter_file.write_text("100") dispatch_mod._sync_counter_if_needed("task-300+1") assert counter_file.read_text().strip() == "301" # --------------------------------------------------------------------------- # 24. TestTaskIdFormatValidation: CLI --task-id 포맷 검증 테스트 # --------------------------------------------------------------------------- class TestTaskIdFormatValidation: """--task-id 포맷 v2 검증 테스트""" def test_valid_simple_format(self, dispatch_mod, monkeypatch, capsys): """task-100 → 경고 없이 통과""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "테스트", "--task-id", "task-100"]) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), patch.object(dispatch_mod.logger, "warning") as mock_warn, ): mock_sub.run.return_value = mock_result dispatch_mod.main() # task-id-format 경고가 없어야 함 format_warnings = [c for c in mock_warn.call_args_list if "task-id-format" in str(c)] assert len(format_warnings) == 0 def test_invalid_format_emits_warning(self, dispatch_mod, monkeypatch, capsys): """task-잘못된형식 → 경고 메시지 출력""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "테스트", "--task-id", "task-잘못된형식"]) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), patch.object(dispatch_mod.logger, "warning") as mock_warn, ): mock_sub.run.return_value = mock_result dispatch_mod.main() format_warnings = [c for c in mock_warn.call_args_list if "task-id-format" in str(c)] assert len(format_warnings) == 1 # --------------------------------------------------------------------------- # 25. _get_busy_bots_info() 테스트 # --------------------------------------------------------------------------- class TestGetBusyBotsInfo: """_get_busy_bots_info() 봇 점유 정보 반환 테스트""" def test_empty_timers_returns_empty(self, dispatch_mod, tmp_path): """빈 task-timers.json이면 빈 딕셔너리 반환""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") result = dispatch_mod._get_busy_bots_info() assert result == {} def test_no_timer_file_returns_empty(self, dispatch_mod, tmp_path): """timer 파일이 없으면 빈 딕셔너리 반환""" timer_file = tmp_path / "memory" / "task-timers.json" if timer_file.exists(): timer_file.unlink() result = dispatch_mod._get_busy_bots_info() assert result == {} def test_dev_team_running_maps_to_bot(self, dispatch_mod, tmp_path): """dev1-team running → bot-b 점유""" timer_file = tmp_path / "memory" / "task-timers.json" data = {"tasks": {"task-1.1": {"team_id": "dev1-team", "status": "running"}}} timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._get_busy_bots_info() assert "bot-b" in result assert result["bot-b"]["task_id"] == "task-1.1" assert result["bot-b"]["team_id"] == "dev1-team" def test_composite_bot_field_mapped(self, dispatch_mod, tmp_path): """composite 태스크의 bot 필드가 반영""" timer_file = tmp_path / "memory" / "task-timers.json" data = {"tasks": {"task-100.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}}} timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._get_busy_bots_info() assert "bot-g" in result assert result["bot-g"]["task_id"] == "task-100.1" assert result["bot-g"]["team_id"] == "composite" def test_completed_tasks_excluded(self, dispatch_mod, tmp_path): """completed 상태는 점유로 간주하지 않음""" timer_file = tmp_path / "memory" / "task-timers.json" data = {"tasks": {"task-1.1": {"team_id": "dev1-team", "status": "completed"}}} timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._get_busy_bots_info() assert result == {} def test_corrupted_json_returns_empty(self, dispatch_mod, tmp_path): """깨진 JSON이면 빈 딕셔너리 반환""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text("INVALID JSON{{{", encoding="utf-8") result = dispatch_mod._get_busy_bots_info() assert result == {} def test_multiple_running_tasks(self, dispatch_mod, tmp_path): """여러 running 태스크가 있을 때 모두 반영""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "dev1-team", "status": "running"}, "task-2.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}, "task-3.1": {"team_id": "dev3-team", "status": "completed"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._get_busy_bots_info() assert "bot-b" in result # dev1-team assert "bot-g" in result # composite assert "bot-d" not in result # dev3-team completed # --------------------------------------------------------------------------- # 24. dev팀 봇 충돌 검사 테스트 (composite 점유 봇 체크) # --------------------------------------------------------------------------- class TestDevTeamBotConflict: """dev팀 dispatch 시 composite/dynamic 작업에 봇이 점유되어 있으면 차단 테스트""" def _make_mock_subprocess(self, returncode=0, stdout=None, stderr=""): mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"status": "ok"}) mock_result.stderr = stderr return mock_result def test_composite_on_bot_g_blocks_dev6_dispatch(self, dispatch_mod, tmp_path): """composite가 bot-g 점유 중 → dev6-team dispatch 시 error""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1220.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev6-team", "테스트 작업", force=False) assert result["status"] == "error" assert "bot-g" in result["message"] assert "composite" in result["message"] assert "task-1220.1" in result["message"] def test_composite_on_bot_b_blocks_dev1_dispatch(self, dispatch_mod, tmp_path): """composite가 bot-b 점유 중 → dev1-team dispatch 시 error""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1220.1": {"team_id": "composite", "status": "running", "bot": "bot-b"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev1-team", "테스트 작업", force=False) assert result["status"] == "error" assert "bot-b" in result["message"] assert "task-1220.1" in result["message"] def test_no_composite_allows_dev6_dispatch(self, dispatch_mod, tmp_path): """composite 없을 때 dev6-team 정상 위임 (회귀 테스트)""" timer_file = tmp_path / "memory" / "task-timers.json" timer_file.write_text(json.dumps({"tasks": {}}), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev6-team", "정상 작업") assert result["status"] == "dispatched" def test_composite_on_bot_g_force_true_allows_dev6(self, dispatch_mod, tmp_path): """composite가 bot-g 점유 중이라도 force=True면 dev6-team dispatch 허용""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1220.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev6-team", "강제 작업", force=True) assert result["status"] == "dispatched" def test_composite_completed_allows_dev6_dispatch(self, dispatch_mod, tmp_path): """composite가 completed면 봇 점유 아님 → dev6-team 정상 dispatch""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1220.1": {"team_id": "composite", "status": "completed", "bot": "bot-g"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev6-team", "정상 작업") assert result["status"] == "dispatched" def test_marketing_on_bot_g_blocks_dev6_dispatch(self, dispatch_mod, tmp_path): """marketing이 bot-g 점유 중 → dev6-team dispatch 시 error""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-500.1": {"team_id": "marketing", "status": "running", "bot": "bot-g"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev6-team", "테스트 작업", force=False) assert result["status"] == "error" assert "bot-g" in result["message"] assert "marketing" in result["message"] def test_conflict_calls_cleanup_task(self, dispatch_mod, tmp_path): """봇 충돌로 거부 시 _cleanup_task가 호출됨""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1220.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") cleanup_called = [] def mock_cleanup(task_id): cleanup_called.append(task_id) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), patch.object(dispatch_mod, "_cleanup_task", side_effect=mock_cleanup), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev6-team", "테스트 작업", force=False) assert result["status"] == "error" assert len(cleanup_called) == 1 def test_same_team_running_not_treated_as_bot_conflict(self, dispatch_mod, tmp_path): """같은 팀(dev6-team)의 running 태스크는 봇 충돌이 아닌 팀 충돌로 처리됨""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-100.1": {"team_id": "dev6-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev6-team", "테스트 작업", force=False) # 이 경우 기존 팀 충돌 검사에서 잡혀야 함 (봇 충돌 아님) assert result["status"] == "error" # 메시지에 "봇"이 아닌 팀 관련 메시지가 포함 assert "같은 팀" in result["message"] or "dev6-team" in result["message"] # --------------------------------------------------------------------------- # 25. _get_busy_bots_info() exclude_task_id 테스트 # --------------------------------------------------------------------------- class TestGetBusyBotsInfoExclude: """_get_busy_bots_info(exclude_task_id=...) 자기 자신 제외 테스트""" def test_exclude_removes_own_entry(self, dispatch_mod, tmp_path): """exclude_task_id로 자기 자신을 제외하면 결과에 포함되지 않음""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-100.1": {"team_id": "dev1-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._get_busy_bots_info(exclude_task_id="task-100.1") assert result == {} def test_exclude_keeps_other_entries(self, dispatch_mod, tmp_path): """exclude_task_id로 자기 자신만 제외, 다른 태스크는 유지""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-100.1": {"team_id": "dev1-team", "status": "running"}, "task-200.1": {"team_id": "composite", "status": "running", "bot": "bot-c"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._get_busy_bots_info(exclude_task_id="task-100.1") assert "bot-c" in result assert result["bot-c"]["task_id"] == "task-200.1" def test_exclude_none_returns_all(self, dispatch_mod, tmp_path): """exclude_task_id=None이면 모든 running 태스크 반환 (기존 동작 호환)""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-100.1": {"team_id": "dev1-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") result = dispatch_mod._get_busy_bots_info() assert "bot-b" in result def test_exclude_prevents_overwrite_of_composite_entry(self, dispatch_mod, tmp_path): """핵심 버그 시나리오: dev팀 자신의 entry가 composite entry를 덮어쓰지 않음""" timer_file = tmp_path / "memory" / "task-timers.json" # composite가 bot-c 점유 + dev2-team도 running (같은 bot-c) data = { "tasks": { "task-1241.1": {"team_id": "composite", "status": "running", "bot": "bot-c"}, "task-1242.1": {"team_id": "dev2-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") # task-1242.1을 제외하면 composite의 bot-c만 남아야 함 result = dispatch_mod._get_busy_bots_info(exclude_task_id="task-1242.1") assert "bot-c" in result assert result["bot-c"]["task_id"] == "task-1241.1" assert result["bot-c"]["team_id"] == "composite" # --------------------------------------------------------------------------- # 26. 봇 충돌 검사: timer 시작 후에도 composite 충돌 감지 (핵심 회귀 테스트) # --------------------------------------------------------------------------- class TestBotConflictWithTimerEntry: """dev팀 dispatch 시 이미 timer가 시작된 상태에서도 composite 봇 충돌 감지""" def _make_mock_subprocess(self, returncode=0, stdout=None, stderr=""): mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"status": "ok"}) mock_result.stderr = stderr return mock_result def test_composite_conflict_detected_despite_own_timer_entry(self, dispatch_mod, tmp_path): """핵심 회귀 테스트: composite가 bot-c 점유 + dev2-team timer entry가 이미 존재해도 충돌 감지""" timer_file = tmp_path / "memory" / "task-timers.json" # 실제 시나리오 재현: timer start가 먼저 실행되어 dev2-team entry가 이미 존재 data = { "tasks": { "task-1241.1": {"team_id": "composite", "status": "running", "bot": "bot-c"}, "task-1242.1": {"team_id": "dev2-team", "status": "running", "description": "테스트"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev2-team", "테스트 작업", task_id="task-1242.1", force=False) assert result["status"] == "error" assert "bot-c" in result["message"] assert "task-1241.1" in result["message"] def test_dynamic_bot_on_dev_team_bot_blocks(self, dispatch_mod, tmp_path): """marketing이 bot-c 점유 + dev2-team dispatch 시 차단""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-500.1": {"team_id": "marketing", "status": "running", "bot": "bot-c"}, "task-501.1": {"team_id": "dev2-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev2-team", "테스트 작업", task_id="task-501.1", force=False) assert result["status"] == "error" assert "bot-c" in result["message"] def test_force_bypasses_conflict_with_timer_entry(self, dispatch_mod, tmp_path): """force=True면 timer entry 존재해도 composite 충돌 무시하고 진행""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1241.1": {"team_id": "composite", "status": "running", "bot": "bot-c"}, "task-1242.1": {"team_id": "dev2-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev2": "key2", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev2-team", "강제 작업", task_id="task-1242.1", force=True) assert result["status"] == "dispatched" # --------------------------------------------------------------------------- # 27. _get_available_bots_with_teams() 헬퍼 함수 테스트 # --------------------------------------------------------------------------- class TestGetAvailableBotsWithTeams: """_get_available_bots_with_teams() — 가용 봇 목록 반환 헬퍼 테스트""" def test_all_bots_available_when_none_busy(self, dispatch_mod): """busy_bots가 빈 dict일 때 8개 봇 전부 반환""" result = dispatch_mod._get_available_bots_with_teams({}) assert len(result) == 8 bot_ids = [entry["bot_id"] for entry in result] for bot in ["bot-b", "bot-c", "bot-d", "bot-e", "bot-f", "bot-g", "bot-h", "bot-i"]: assert bot in bot_ids def test_excludes_busy_bots(self, dispatch_mod): """bot-b, bot-g가 busy일 때 나머지 6개만 반환""" busy_bots = { "bot-b": {"team_id": "composite", "task_id": "task-1.1"}, "bot-g": {"team_id": "composite", "task_id": "task-2.1"}, } result = dispatch_mod._get_available_bots_with_teams(busy_bots) assert len(result) == 6 bot_ids = [entry["bot_id"] for entry in result] assert "bot-b" not in bot_ids assert "bot-g" not in bot_ids for bot in ["bot-c", "bot-d", "bot-e", "bot-f", "bot-h", "bot-i"]: assert bot in bot_ids def test_all_bots_busy_returns_empty(self, dispatch_mod): """모든 봇이 busy일 때 빈 리스트 반환""" busy_bots = { "bot-b": {"team_id": "composite", "task_id": "task-1.1"}, "bot-c": {"team_id": "composite", "task_id": "task-2.1"}, "bot-d": {"team_id": "composite", "task_id": "task-3.1"}, "bot-e": {"team_id": "composite", "task_id": "task-4.1"}, "bot-f": {"team_id": "composite", "task_id": "task-5.1"}, "bot-g": {"team_id": "composite", "task_id": "task-6.1"}, "bot-h": {"team_id": "composite", "task_id": "task-7.1"}, "bot-i": {"team_id": "composite", "task_id": "task-8.1"}, } result = dispatch_mod._get_available_bots_with_teams(busy_bots) assert result == [] def test_returns_correct_default_team_mapping(self, dispatch_mod): """각 봇의 default_team이 TEAM_TO_BOT_ID 역매핑과 일치""" expected_mapping = { "bot-b": "dev1-team", "bot-c": "dev2-team", "bot-d": "dev3-team", "bot-e": "dev4-team", "bot-f": "dev5-team", "bot-g": "dev6-team", "bot-h": "dev7-team", "bot-i": "dev8-team", } result = dispatch_mod._get_available_bots_with_teams({}) for entry in result: bot_id = entry["bot_id"] assert entry["default_team"] == expected_mapping[bot_id] # --------------------------------------------------------------------------- # 28. 봇 충돌 에러 응답에 가용 봇 추천 포함 테스트 # --------------------------------------------------------------------------- class TestBotConflictAvailableBots: """봇 충돌 에러 응답에 가용 봇 추천이 포함되는지 테스트""" def _make_mock_subprocess(self, returncode=0, stdout=None, stderr=""): mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"status": "ok"}) mock_result.stderr = stderr return mock_result def test_conflict_error_includes_available_bots_field(self, dispatch_mod, tmp_path): """composite가 bot-g 점유 중 → dev6-team dispatch error에 available_bots 필드 존재""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1220.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev6-team", "테스트 작업", force=False) assert result["status"] == "error" assert "available_bots" in result def test_conflict_error_message_includes_alternatives(self, dispatch_mod, tmp_path): """에러 메시지에 '가용 대안:' 문자열과 가용 봇 정보 포함""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1220.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev6-team", "테스트 작업", force=False) assert result["status"] == "error" assert "가용 대안:" in result["message"] def test_conflict_error_available_bots_excludes_busy(self, dispatch_mod, tmp_path): """available_bots에 점유 중인 봇(bot-g)이 포함되지 않음""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1220.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev6-team", "테스트 작업", force=False) assert result["status"] == "error" available_bot_ids = [entry["bot_id"] for entry in result.get("available_bots", [])] assert "bot-g" not in available_bot_ids def test_conflict_all_bots_busy_message(self, dispatch_mod, tmp_path): """모든 봇이 busy일 때 '모든 봇이 작업 중' 메시지""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "composite", "status": "running", "bot": "bot-b"}, "task-2.1": {"team_id": "composite", "status": "running", "bot": "bot-c"}, "task-3.1": {"team_id": "composite", "status": "running", "bot": "bot-d"}, "task-4.1": {"team_id": "composite", "status": "running", "bot": "bot-e"}, "task-5.1": {"team_id": "composite", "status": "running", "bot": "bot-f"}, "task-6.1": {"team_id": "composite", "status": "running", "bot": "bot-g"}, "task-7.1": {"team_id": "composite", "status": "running", "bot": "bot-h"}, "task-8.1": {"team_id": "composite", "status": "running", "bot": "bot-i"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev6": "key6", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev6-team", "테스트 작업", force=False) assert result["status"] == "error" assert "모든 봇이 작업 중" in result["message"] # --------------------------------------------------------------------------- # 29. 이미지 QC 게이트 강제 차단 테스트 (task-1321.1) # --------------------------------------------------------------------------- class TestImageQcGateBlock: """이미지/광고 작업 시 --workflow 미적용 시 차단 + --skip-qc-gate 우회 테스트""" def test_image_keyword_without_workflow_exits(self, dispatch_mod, monkeypatch): """이미지 키워드 포함 + --workflow 없음 → sys.exit(1) 호출""" monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "배너 이미지 생성"]) with pytest.raises(SystemExit) as exc_info: dispatch_mod.main() assert exc_info.value.code == 1 def test_image_keyword_with_skip_qc_gate_passes(self, dispatch_mod, monkeypatch, capsys): """이미지 키워드 + --skip-qc-gate → 차단 없이 통과 (dispatch 진행)""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" monkeypatch.setattr( sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "배너 이미지 생성", "--skip-qc-gate"], ) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert "status" in output def test_image_keyword_with_workflow_passes(self, dispatch_mod, monkeypatch, capsys): """이미지 키워드 + --workflow image-qc-gate → 차단 없이 통과""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" monkeypatch.setattr( sys, "argv", [ "dispatch.py", "--team", "dev1-team", "--task", "배너 이미지 생성", "--workflow", "image-qc-gate", ], ) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), patch("prompts.image_workflow.build_workflow_overview_prompt", return_value="workflow prompt"), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert "status" in output def test_no_image_keyword_without_workflow_passes(self, dispatch_mod, monkeypatch, capsys): """이미지 키워드 없음 + --workflow 없음 → 정상 통과 (영향 없음)""" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"status": "ok"}) mock_result.stderr = "" monkeypatch.setattr( sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "API 엔드포인트 구현"], ) with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = mock_result dispatch_mod.main() captured = capsys.readouterr() output = json.loads(captured.out) assert "status" in output def test_english_banner_keyword_blocks(self, dispatch_mod, monkeypatch): """영문 'banner' 키워드도 차단""" monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "Create banner design"]) with pytest.raises(SystemExit) as exc_info: dispatch_mod.main() assert exc_info.value.code == 1 def test_english_image_keyword_blocks(self, dispatch_mod, monkeypatch): """영문 'image' 키워드도 차단""" monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "Generate product image"]) with pytest.raises(SystemExit) as exc_info: dispatch_mod.main() assert exc_info.value.code == 1 def test_korean_ad_keyword_blocks(self, dispatch_mod, monkeypatch): """한국어 '광고' 키워드도 차단""" monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "메타 광고 소재 제작"]) with pytest.raises(SystemExit) as exc_info: dispatch_mod.main() assert exc_info.value.code == 1 def test_korean_design_keyword_blocks(self, dispatch_mod, monkeypatch): """한국어 '디자인' 키워드도 차단""" monkeypatch.setattr(sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", "UI 디자인 작업"]) with pytest.raises(SystemExit) as exc_info: dispatch_mod.main() assert exc_info.value.code == 1 # --------------------------------------------------------------------------- # XX. 논리적팀-dev팀 봇 충돌 방지 테스트 (task-1405.1) # --------------------------------------------------------------------------- class TestLogicalTeamBotConflict: """논리적팀(design/marketing/content)이 봇 점유 시 해당 봇의 dev팀 dispatch 차단 테스트""" def _make_mock_subprocess(self, returncode=0, stdout=None, stderr=""): mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"status": "ok"}) mock_result.stderr = stderr return mock_result def test_design_on_bot_e_blocks_dev4_dispatch(self, dispatch_mod, tmp_path): """design이 bot-e 점유 중 → dev4-team dispatch 시 error""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1400.1": {"team_id": "design", "status": "running", "bot": "bot-e"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev4": "key4", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev4-team", "테스트 작업", force=False) assert result["status"] == "error" assert "bot-e" in result["message"] assert "design" in result["message"] assert "task-1400.1" in result["message"] def test_content_on_bot_b_blocks_dev1_dispatch(self, dispatch_mod, tmp_path): """content가 bot-b 점유 중 → dev1-team dispatch 시 error""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1380.1": {"team_id": "content", "status": "running", "bot": "bot-b"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0) result = dispatch_mod.dispatch("dev1-team", "테스트 작업", force=False) assert result["status"] == "error" assert "bot-b" in result["message"] assert "content" in result["message"] assert "task-1380.1" in result["message"] def test_design_completed_allows_dev4_dispatch(self, dispatch_mod, tmp_path): """design이 completed면 봇 점유 아님 → dev4-team 정상 dispatch""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1400.1": {"team_id": "design", "status": "completed", "bot": "bot-e"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev4": "key4", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev4-team", "정상 작업") assert result["status"] == "dispatched" def test_logical_team_without_bot_field_no_false_block(self, dispatch_mod, tmp_path): """논리적팀이 running이지만 bot 필드 없음 → dev팀 정상 dispatch (false block 없음)""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1400.1": {"team_id": "design", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev1": "key1", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev1-team", "정상 작업") assert result["status"] == "dispatched" def test_multiple_logical_teams_different_bots(self, dispatch_mod, tmp_path): """여러 논리적팀이 각각 다른 봇 점유 → 해당 dev팀만 차단, 다른 팀은 정상""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "design", "status": "running", "bot": "bot-e"}, "task-2.1": {"team_id": "marketing", "status": "running", "bot": "bot-b"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with ( patch.object(dispatch_mod, "subprocess") as mock_sub, patch.object(dispatch_mod, "BOT_KEYS", {"dev3": "key3", "anu": "anu-key"}), ): mock_sub.run.return_value = self._make_mock_subprocess(0, json.dumps({"status": "ok"})) result = dispatch_mod.dispatch("dev3-team", "정상 작업") assert result["status"] == "dispatched" # --------------------------------------------------------------------------- # XX. _select_and_reserve_bot() 원자적 봇 선택 + 예약 테스트 (task-1405.1) # --------------------------------------------------------------------------- class TestSelectAndReserveBot: """_select_and_reserve_bot() 원자적 봇 선택 + 예약 테스트""" def test_selects_first_available_bot(self, dispatch_mod, tmp_path): """빈 timers → bot-b(첫 번째 봇) 선택 + timer에 bot 필드 기록 확인""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-test.1": {"team_id": "dev1-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") selected = dispatch_mod._select_and_reserve_bot("task-test.1") assert selected == "bot-b" # timer 파일에 bot 필드가 기록되었는지 확인 written = json.loads(timer_file.read_text(encoding="utf-8")) assert written["tasks"]["task-test.1"].get("bot") == "bot-b" def test_skips_busy_bots(self, dispatch_mod, tmp_path): """dev1(bot-b), dev2(bot-c) running → bot-d 선택""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-other1.1": {"team_id": "dev1-team", "status": "running"}, "task-other2.1": {"team_id": "dev2-team", "status": "running"}, "task-test.1": {"team_id": "dev3-team", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") selected = dispatch_mod._select_and_reserve_bot("task-test.1") assert selected == "bot-d" def test_reserves_bot_in_timer_file(self, dispatch_mod, tmp_path): """봇 선택 후 task-timers.json을 다시 읽어서 해당 task의 bot 필드가 설정되었는지 확인""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-reserve.1": {"team_id": "dev1-team", "status": "running", "description": "예약 테스트"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") selected = dispatch_mod._select_and_reserve_bot("task-reserve.1") # 파일을 다시 읽어서 bot 필드 확인 written = json.loads(timer_file.read_text(encoding="utf-8")) task_entry = written["tasks"]["task-reserve.1"] assert "bot" in task_entry assert task_entry["bot"] == selected def test_raises_when_all_busy(self, dispatch_mod, tmp_path): """모든 봇이 busy → RuntimeError 발생""" timer_file = tmp_path / "memory" / "task-timers.json" data = { "tasks": { "task-1.1": {"team_id": "dev1-team", "status": "running"}, "task-2.1": {"team_id": "dev2-team", "status": "running"}, "task-3.1": {"team_id": "dev3-team", "status": "running"}, "task-4.1": {"team_id": "dev4-team", "status": "running"}, "task-5.1": {"team_id": "dev5-team", "status": "running"}, "task-6.1": {"team_id": "dev6-team", "status": "running"}, "task-7.1": {"team_id": "dev7-team", "status": "running"}, "task-8.1": {"team_id": "dev8-team", "status": "running"}, "task-test.1": {"team_id": "composite", "status": "running"}, } } timer_file.write_text(json.dumps(data), encoding="utf-8") with pytest.raises(RuntimeError): dispatch_mod._select_and_reserve_bot("task-test.1") # --------------------------------------------------------------------------- # XX. TestValidateModelConsistency: _validate_model_consistency() 테스트 # --------------------------------------------------------------------------- class TestValidateModelConsistency: """_validate_model_consistency() 단위 테스트""" def _make_org_fixture(self, tmp_path, org_model: str = "claude-opus-4-6") -> "Path": org_file = tmp_path / "org-structure.json" data = { "structure": { "columns": { "teams": [ { "team_id": "development-office", "sub_teams": [ { "sub_team_id": "dev3-team", "lead": {"id": "dagda", "model": org_model}, } ], }, { "team_id": "marketing-team", "lead": {"id": "aphrodite", "model": "claude-opus-4-6"}, }, ] } } } org_file.write_text(json.dumps(data), encoding="utf-8") return org_file def _make_bot_settings_fixture( self, tmp_path, bot_model: str = "claude-opus-4-6", key_hash: str = "0b94683120a691cf" ) -> "Path": settings_file = tmp_path / "bot_settings.json" data = { key_hash: { "display_name": "dev3_Dagda_bot", "models": {_TEST_CHAT_ID: bot_model}, "token": "fake-token-12345", } } settings_file.write_text(json.dumps(data), encoding="utf-8") return settings_file def test_consistent_models(self, tmp_path, monkeypatch): """org와 bot 모델이 같을 때 consistent=True""" org_file = self._make_org_fixture(tmp_path, org_model="claude-opus-4-6") settings_file = self._make_bot_settings_fixture(tmp_path, bot_model="claude-opus-4-6") monkeypatch.setattr("dispatch.ORG_FILE", org_file) monkeypatch.setattr("dispatch.TEAM_BOT", {"dev3-team": "dev3"}) monkeypatch.setattr("dispatch.BOT_KEYS", {"dev3": "0b94683120a691cf"}) monkeypatch.setattr("dispatch.CHAT_ID", _TEST_CHAT_ID) monkeypatch.setattr( "dispatch.Path", lambda *args, **kwargs: settings_file.parent if args == () else __import__("pathlib").Path(*args, **kwargs), ) # Path.home() 패치: bot_settings.json 경로를 tmp_path로 유도 import pathlib class _FakeHome: def __truediv__(self, other): return _FakeCokacdir(tmp_path) class _FakeCokacdir: def __init__(self, base): self._base = base def __truediv__(self, other): return self._base / other original_home = pathlib.Path.home monkeypatch.setattr(pathlib.Path, "home", staticmethod(lambda: _FakeHome())) import importlib import sys if "dispatch" in sys.modules: del sys.modules["dispatch"] workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) import dispatch as dispatch_mod monkeypatch.setattr(dispatch_mod, "ORG_FILE", org_file) monkeypatch.setattr(dispatch_mod, "TEAM_BOT", {"dev3-team": "dev3"}) monkeypatch.setattr(dispatch_mod, "BOT_KEYS", {"dev3": "0b94683120a691cf"}) monkeypatch.setattr(dispatch_mod, "CHAT_ID", _TEST_CHAT_ID) result = dispatch_mod._validate_model_consistency("dev3-team") assert result["consistent"] is True assert result["org_model"] == "claude-opus-4-6" assert result["bot_model"] == "claude-opus-4-6" assert result["team_id"] == "dev3-team" monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) def test_inconsistent_models(self, tmp_path, monkeypatch, caplog): """org와 bot 모델이 다를 때 consistent=False + WARNING 로그""" import logging import pathlib org_file = self._make_org_fixture(tmp_path, org_model="claude-opus-4-6") settings_file = self._make_bot_settings_fixture(tmp_path, bot_model="claude-sonnet-4-5") class _FakeHome: def __truediv__(self, other): return _FakeCokacdir(tmp_path) class _FakeCokacdir: def __init__(self, base): self._base = base def __truediv__(self, other): return self._base / other original_home = pathlib.Path.home monkeypatch.setattr(pathlib.Path, "home", staticmethod(lambda: _FakeHome())) import sys workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) if "dispatch" in sys.modules: del sys.modules["dispatch"] import dispatch as dispatch_mod monkeypatch.setattr(dispatch_mod, "ORG_FILE", org_file) monkeypatch.setattr(dispatch_mod, "TEAM_BOT", {"dev3-team": "dev3"}) monkeypatch.setattr(dispatch_mod, "BOT_KEYS", {"dev3": "0b94683120a691cf"}) monkeypatch.setattr(dispatch_mod, "CHAT_ID", _TEST_CHAT_ID) with caplog.at_level(logging.WARNING): result = dispatch_mod._validate_model_consistency("dev3-team") assert result["consistent"] is False assert result["org_model"] == "claude-opus-4-6" assert result["bot_model"] == "claude-sonnet-4-5" warning_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING] assert any( "모델 불일치" in msg for msg in warning_messages ), f"WARNING 로그에 '모델 불일치' 미포함. 기록된 로그: {warning_messages}" monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) def test_org_file_missing(self, tmp_path, monkeypatch): """org-structure.json 없을 때 graceful (에러 안남)""" import pathlib import sys missing_org = tmp_path / "nonexistent-org.json" settings_file = self._make_bot_settings_fixture(tmp_path) class _FakeHome: def __truediv__(self, other): return _FakeCokacdir(tmp_path) class _FakeCokacdir: def __init__(self, base): self._base = base def __truediv__(self, other): return self._base / other original_home = pathlib.Path.home monkeypatch.setattr(pathlib.Path, "home", staticmethod(lambda: _FakeHome())) workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) if "dispatch" in sys.modules: del sys.modules["dispatch"] import dispatch as dispatch_mod monkeypatch.setattr(dispatch_mod, "ORG_FILE", missing_org) monkeypatch.setattr(dispatch_mod, "TEAM_BOT", {"dev3-team": "dev3"}) monkeypatch.setattr(dispatch_mod, "BOT_KEYS", {"dev3": "0b94683120a691cf"}) monkeypatch.setattr(dispatch_mod, "CHAT_ID", _TEST_CHAT_ID) # 에러 없이 실행되어야 함 result = dispatch_mod._validate_model_consistency("dev3-team") assert isinstance(result, dict) assert "consistent" in result assert result["team_id"] == "dev3-team" # org 파일 없으면 org_model은 빈 문자열 → consistent=True (확인 불가) assert result["org_model"] == "" monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) def test_bot_settings_missing(self, tmp_path, monkeypatch): """bot_settings.json 없을 때 graceful (에러 안남)""" import pathlib import sys org_file = self._make_org_fixture(tmp_path) # bot_settings.json을 존재하지 않는 경로로 유도 empty_dir = tmp_path / "empty_home" empty_dir.mkdir() class _FakeHome: def __truediv__(self, other): return _FakeCokacdir(empty_dir) class _FakeCokacdir: def __init__(self, base): self._base = base def __truediv__(self, other): return self._base / other original_home = pathlib.Path.home monkeypatch.setattr(pathlib.Path, "home", staticmethod(lambda: _FakeHome())) workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) if "dispatch" in sys.modules: del sys.modules["dispatch"] import dispatch as dispatch_mod monkeypatch.setattr(dispatch_mod, "ORG_FILE", org_file) monkeypatch.setattr(dispatch_mod, "TEAM_BOT", {"dev3-team": "dev3"}) monkeypatch.setattr(dispatch_mod, "BOT_KEYS", {"dev3": "0b94683120a691cf"}) monkeypatch.setattr(dispatch_mod, "CHAT_ID", _TEST_CHAT_ID) # 에러 없이 실행되어야 함 result = dispatch_mod._validate_model_consistency("dev3-team") assert isinstance(result, dict) assert "consistent" in result assert result["team_id"] == "dev3-team" assert result["bot_model"] == "" monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) def test_unknown_team_id(self, tmp_path, monkeypatch): """존재하지 않는 team_id 전달 시 기본값 반환""" import pathlib import sys org_file = self._make_org_fixture(tmp_path) settings_file = self._make_bot_settings_fixture(tmp_path) class _FakeHome: def __truediv__(self, other): return _FakeCokacdir(tmp_path) class _FakeCokacdir: def __init__(self, base): self._base = base def __truediv__(self, other): return self._base / other original_home = pathlib.Path.home monkeypatch.setattr(pathlib.Path, "home", staticmethod(lambda: _FakeHome())) workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) if "dispatch" in sys.modules: del sys.modules["dispatch"] import dispatch as dispatch_mod monkeypatch.setattr(dispatch_mod, "ORG_FILE", org_file) monkeypatch.setattr(dispatch_mod, "TEAM_BOT", {"dev3-team": "dev3"}) monkeypatch.setattr(dispatch_mod, "BOT_KEYS", {"dev3": "0b94683120a691cf"}) monkeypatch.setattr(dispatch_mod, "CHAT_ID", _TEST_CHAT_ID) # 존재하지 않는 팀 ID — TEAM_BOT에 없어서 bot_model="" result = dispatch_mod._validate_model_consistency("nonexistent-team") assert isinstance(result, dict) assert result["team_id"] == "nonexistent-team" # org에도 없고 bot도 없으면 둘 다 빈 문자열 → consistent=True (비교 불가) assert result["consistent"] is True assert result["org_model"] == "" assert result["bot_model"] == "" monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) # --------------------------------------------------------------------------- # XX. TestSyncBotSettings: _sync_bot_settings() 테스트 # --------------------------------------------------------------------------- class TestSyncBotSettings: """_sync_bot_settings() 단위 테스트""" def _make_bot_settings(self, home_dir: "Path", key_hash: str = "0b94683120a691cf") -> "Path": cokacdir = home_dir / ".cokacdir" cokacdir.mkdir(parents=True, exist_ok=True) settings_file = cokacdir / "bot_settings.json" data = { key_hash: { "display_name": "dev3_Dagda_bot", "models": {_TEST_CHAT_ID: "claude-opus-4-6"}, "token": "fake-token-12345", } } settings_file.write_text(json.dumps(data), encoding="utf-8") return settings_file def _patch_env(self, tmp_path, monkeypatch): """공통 monkeypatch: WORKSPACE와 Path.home()을 tmp_path 기반으로 교체""" import pathlib import sys workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) if "dispatch" in sys.modules: del sys.modules["dispatch"] import dispatch as dispatch_mod # WORKSPACE를 tmp_path로 교체 (tmp_path / "memory").mkdir(parents=True, exist_ok=True) monkeypatch.setattr(dispatch_mod, "WORKSPACE", tmp_path) # Path.home()을 tmp_path로 유도 original_home = pathlib.Path.home fake_home = tmp_path / "fake_home" fake_home.mkdir(parents=True, exist_ok=True) monkeypatch.setattr(pathlib.Path, "home", staticmethod(lambda: fake_home)) return dispatch_mod, fake_home, original_home def test_sync_creates_file(self, tmp_path, monkeypatch): """동기화 파일이 올바르게 생성됨""" dispatch_mod, fake_home, original_home = self._patch_env(tmp_path, monkeypatch) import pathlib self._make_bot_settings(fake_home) dispatch_mod._sync_bot_settings() sync_path = tmp_path / "memory" / "bot_settings_sync.json" assert sync_path.exists(), f"동기화 파일이 생성되지 않음: {sync_path}" monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) def test_token_masked(self, tmp_path, monkeypatch): """token 값이 ***REDACTED***로 교체됨""" dispatch_mod, fake_home, original_home = self._patch_env(tmp_path, monkeypatch) import pathlib self._make_bot_settings(fake_home) dispatch_mod._sync_bot_settings() sync_path = tmp_path / "memory" / "bot_settings_sync.json" synced = json.loads(sync_path.read_text(encoding="utf-8")) for key_hash, cfg in synced.items(): assert cfg.get("token") == "***REDACTED***", f"token이 마스킹되지 않음: {cfg.get('token')}" monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) def test_other_fields_preserved(self, tmp_path, monkeypatch): """token 외 필드는 그대로 유지됨""" dispatch_mod, fake_home, original_home = self._patch_env(tmp_path, monkeypatch) import pathlib self._make_bot_settings(fake_home, key_hash="0b94683120a691cf") dispatch_mod._sync_bot_settings() sync_path = tmp_path / "memory" / "bot_settings_sync.json" synced = json.loads(sync_path.read_text(encoding="utf-8")) cfg = synced.get("0b94683120a691cf", {}) assert cfg.get("display_name") == "dev3_Dagda_bot" assert cfg.get("models") == {_TEST_CHAT_ID: "claude-opus-4-6"} monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) def test_bot_settings_missing(self, tmp_path, monkeypatch): """원본 bot_settings.json 없을 때 graceful (에러 안남)""" dispatch_mod, fake_home, original_home = self._patch_env(tmp_path, monkeypatch) import pathlib # bot_settings.json을 생성하지 않음 (fake_home/.cokacdir/bot_settings.json 없음) # 에러 없이 실행되어야 하며, sync 파일도 생성되지 않아야 함 dispatch_mod._sync_bot_settings() sync_path = tmp_path / "memory" / "bot_settings_sync.json" assert not sync_path.exists(), "원본 없는데 동기화 파일이 생성됨" monkeypatch.setattr(pathlib.Path, "home", staticmethod(original_home)) class TestPrdDecomposition: """PRD 자동 분해 (--prd) 기능 테스트""" def test_parse_prd_regex_basic(self): """기본 Phase 추출""" prd_content = """ ## 3. 구현 로드맵 ### Phase 1 (2일) — 기초 구현 [F1, F2] - 항목 1 - 항목 2 - **DoD**: 기초 구현 완료 ### Phase 2 (1일) — 통합 [F3] - 통합 테스트 """ from dispatch import _parse_prd_regex phases = _parse_prd_regex(prd_content) assert len(phases) == 2 assert phases[0]["phase_type"] == "Phase" assert phases[0]["phase_number"] == 1 assert phases[0]["title"] == "기초 구현 [F1, F2]" assert phases[0]["duration"] == "2일" assert phases[0]["features"] == ["F1", "F2"] assert phases[0]["dod"] is not None assert "기초 구현 완료" in phases[0]["dod"] assert phases[1]["phase_number"] == 2 def test_parse_prd_regex_with_sprint(self): """Sprint 0 포함 추출""" prd_content = """ ### Sprint 0 (0.5일) — 준비 - 준비 항목 ### Phase 1 (2일) — 구현 - 구현 항목 """ from dispatch import _parse_prd_regex phases = _parse_prd_regex(prd_content) assert len(phases) == 2 assert phases[0]["phase_type"] == "Sprint" assert phases[0]["phase_number"] == 0 assert phases[1]["phase_type"] == "Phase" def test_parse_prd_regex_empty(self): """Phase 없는 문서""" from dispatch import _parse_prd_regex phases = _parse_prd_regex("# 빈 문서\n내용만 있음") assert phases == [] def test_handle_prd_nonexistent_file(self): """존재하지 않는 PRD 파일 에러""" from dispatch import handle_prd result = handle_prd("/nonexistent/path.md", "dev1-team") assert result["status"] == "error" assert "찾을 수 없습니다" in result["message"] def test_handle_prd_creates_files(self, tmp_path): """task 파일 생성 확인""" import os from unittest.mock import patch from dispatch import handle_prd prd_content = """ ## 3. 구현 로드맵 ### Phase 1 (2일) — 기초 구현 [F1, F2] - 항목 1 - **DoD**: 구현 완료 ### Phase 2 (1일) — 통합 - 통합 항목 """ # 테스트용 PRD 파일 생성 prd_file = tmp_path / "prd-test-project.md" prd_file.write_text(prd_content, encoding="utf-8") # WORKSPACE를 tmp_path로 패치하여 실제 task 디렉토리에 영향 주지 않음 tasks_dir = tmp_path / "memory" / "tasks" tasks_dir.mkdir(parents=True, exist_ok=True) with patch("dispatch.WORKSPACE", tmp_path): result = handle_prd(str(prd_file), "dev1-team") assert result["status"] == "ok" assert result["method"] == "regex" assert result["total_phases"] == 2 assert len(result["created"]) == 2 assert len(result["skipped"]) == 0 # 생성된 파일 내용 검증 phase1_file = tasks_dir / "dispatch-test-project-phase1.md" assert phase1_file.exists() content = phase1_file.read_text(encoding="utf-8") assert "task_id:" in content assert "dev1-team" in content assert "F1, F2" in content def test_handle_prd_skips_existing(self, tmp_path): """기존 파일 스킵 확인""" from unittest.mock import patch from dispatch import handle_prd prd_content = """ ### Phase 1 (1일) — 테스트 - 항목 """ prd_file = tmp_path / "prd-skip-test.md" prd_file.write_text(prd_content, encoding="utf-8") tasks_dir = tmp_path / "memory" / "tasks" tasks_dir.mkdir(parents=True, exist_ok=True) # 이미 존재하는 파일 생성 existing = tasks_dir / "dispatch-skip-test-phase1.md" existing.write_text("existing", encoding="utf-8") with patch("dispatch.WORKSPACE", tmp_path): result = handle_prd(str(prd_file), "dev1-team") assert result["status"] == "ok" assert len(result["created"]) == 0 assert len(result["skipped"]) == 1 # 기존 내용이 보존되었는지 확인 assert existing.read_text() == "existing" class TestWakeUpDelay: """task-2046: wake-up 실패 시 dispatch delay가 5초로 설정되는지 검증""" def test_wake_up_failure_sets_delay_to_5(self): """봇 프로세스 미감지 + wake-up 실패 시 _dispatch_delay = 5""" import dispatch as mod # _wake_up_bot이 False를 반환하는 시나리오 original_check = mod._check_bot_process original_wake = mod._wake_up_bot try: mod._check_bot_process = lambda key: False mod._wake_up_bot = lambda chat_id, key: False # wake-up 로직 시뮬레이션 _dispatch_delay = 10 if not mod._check_bot_process("test_key"): _wake_result = mod._wake_up_bot("chat", "test_key") if not _wake_result: _dispatch_delay = 5 assert _dispatch_delay == 5, f"Expected 5, got {_dispatch_delay}" finally: mod._check_bot_process = original_check mod._wake_up_bot = original_wake def test_bot_already_running_keeps_default_delay(self): """봇 프로세스 감지 시 기본 딜레이 10초 유지""" import dispatch as mod original_check = mod._check_bot_process try: mod._check_bot_process = lambda key: True _dispatch_delay = 10 if not mod._check_bot_process("test_key"): _dispatch_delay = 5 assert _dispatch_delay == 10, f"Expected 10, got {_dispatch_delay}" finally: mod._check_bot_process = original_check