""" test_chain.py chain.py Phase 자동 체이닝 시스템 단위 테스트 (아르고스 작성) 테스트 항목: - TestCreate: 체인 파일 생성, 중복 ID 에러 - TestAddPhase: Phase 추가, tasks 스키마 정규화, 잘못된 JSON 에러 - TestTaskDone: 완료 마킹, 미완료 대기, Phase 전환, paused 무시 - TestStatus: JSON 출력 확인 - TestList: 빈 목록, 여러 체인 목록 - TestUpdateChainTask: dispatch.py _update_chain_task 함수 동작 """ import argparse import json import os import subprocess import sys import types from datetime import datetime from pathlib import Path from unittest.mock import MagicMock, patch import pytest # --------------------------------------------------------------------------- # 헬퍼: chain 모듈을 격리된 WORKSPACE로 재로드 # --------------------------------------------------------------------------- def _load_chain_with_workspace(tmp_path: Path) -> types.ModuleType: """chain 모듈을 tmp_path를 WORKSPACE로 설정하여 로드한다. chain.py는 모듈 최상단에서 WORKSPACE를 환경변수 기준으로 결정하므로 sys.modules에서 제거 후 WORKSPACE 패치를 적용한다. """ workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) for mod_name in list(sys.modules.keys()): if mod_name == "chain": del sys.modules[mod_name] import chain as _chain _chain.WORKSPACE = tmp_path _chain.CHAINS_DIR = tmp_path / "memory" / "chains" return _chain def _load_dispatch_with_workspace(tmp_path: Path) -> types.ModuleType: """dispatch 모듈을 tmp_path를 WORKSPACE로 설정하여 로드한다.""" workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(workspace) not in sys.path: sys.path.insert(0, str(workspace)) import prompts.team_prompts # noqa: F401 for mod_name in list(sys.modules.keys()): if mod_name == "dispatch": del sys.modules[mod_name] import dispatch as _dispatch _dispatch.WORKSPACE = tmp_path return _dispatch # --------------------------------------------------------------------------- # 헬퍼: task-done 테스트용 체인 데이터 셋업 # --------------------------------------------------------------------------- def _setup_chain_with_phases(tmp_path, chain_id="test-chain"): """2개 Phase, Phase 0에 2개 in_progress task, Phase 1에 1개 pending task를 생성한다.""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) data = { "chain_id": chain_id, "description": "Test", "status": "active", "current_phase_idx": 0, "created_at": datetime.now().isoformat(), "phases": [ { "name": "Phase 1", "status": "in_progress", "tasks": [ { "team": "dev1-team", "task_id": "task-1.1", "description": "작업A", "level": "normal", "status": "in_progress", "dispatched_at": "2026-01-01", "completed_at": None, }, { "team": "dev2-team", "task_id": "task-2.1", "description": "작업B", "level": "normal", "status": "in_progress", "dispatched_at": "2026-01-01", "completed_at": None, }, ], }, { "name": "Phase 2", "status": "pending", "tasks": [ { "team": "dev1-team", "task_id": None, "description": "작업C", "level": "normal", "status": "pending", "dispatched_at": None, "completed_at": None, }, ], }, ], } chain_file = chains_dir / f"{chain_id}.json" chain_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") return data # --------------------------------------------------------------------------- # fixture: tmp_path 기반 chain 모듈 # --------------------------------------------------------------------------- @pytest.fixture() def chain_mod(tmp_path): """격리된 WORKSPACE를 사용하는 chain 모듈을 반환한다.""" (tmp_path / "memory" / "chains").mkdir(parents=True, exist_ok=True) return _load_chain_with_workspace(tmp_path) @pytest.fixture() def dispatch_mod(tmp_path): """격리된 WORKSPACE를 사용하는 dispatch 모듈을 반환한다.""" (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) yield mod # 테스트 후 WORKSPACE를 실제 경로로 복원하여 다른 테스트 격리 mod.WORKSPACE = real_workspace # sys.modules["dispatch"]를 원래 모듈로 복원 if _original_dispatch is not None: sys.modules["dispatch"] = _original_dispatch # --------------------------------------------------------------------------- # 1. TestCreate: 체인 파일 생성, 중복 ID 에러 # --------------------------------------------------------------------------- class TestCreate: """cmd_create() 서브커맨드 테스트""" def test_create_chain_file(self, chain_mod, tmp_path): """cmd_create로 체인 파일이 생성되고 필수 필드가 올바르게 설정된다.""" args = argparse.Namespace(id="my-chain", desc="테스트 체인") chain_mod.cmd_create(args) chain_file = tmp_path / "memory" / "chains" / "my-chain.json" assert chain_file.exists() data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["chain_id"] == "my-chain" assert data["description"] == "테스트 체인" assert data["status"] == "active" assert data["current_phase_idx"] == 0 assert data["phases"] == [] def test_create_prints_ok_message(self, chain_mod, tmp_path, capsys): """cmd_create 성공 시 stdout에 [OK] 메시지가 출력된다.""" args = argparse.Namespace(id="ok-chain", desc="OK 메시지 테스트") chain_mod.cmd_create(args) captured = capsys.readouterr() assert "[OK]" in captured.out def test_create_duplicate_exits(self, chain_mod, tmp_path): """같은 ID로 두 번 생성 시 sys.exit(1)이 발생한다.""" args = argparse.Namespace(id="dup-chain", desc="중복 테스트") chain_mod.cmd_create(args) with pytest.raises(SystemExit) as exc_info: chain_mod.cmd_create(args) assert exc_info.value.code == 1 # --------------------------------------------------------------------------- # 2. TestAddPhase: Phase 추가, tasks 스키마 정규화, 잘못된 JSON 에러 # --------------------------------------------------------------------------- class TestAddPhase: """cmd_add_phase() 서브커맨드 테스트""" def _create_chain(self, chain_mod, chain_id="base-chain"): """테스트용 체인 파일을 생성하는 헬퍼.""" args = argparse.Namespace(id=chain_id, desc="베이스 체인") chain_mod.cmd_create(args) def test_add_phase_normal(self, chain_mod, tmp_path): """Phase 추가 후 phases 배열에 정확히 1개 항목이 추가된다.""" self._create_chain(chain_mod) tasks_json = json.dumps([{"team": "dev1-team", "desc": "작업 설명"}]) args = argparse.Namespace(chain="base-chain", name="Phase 1", tasks=tasks_json) chain_mod.cmd_add_phase(args) chain_file = tmp_path / "memory" / "chains" / "base-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) assert len(data["phases"]) == 1 assert data["phases"][0]["name"] == "Phase 1" assert len(data["phases"][0]["tasks"]) == 1 def test_tasks_normalization(self, chain_mod, tmp_path): """task 스키마 정규화: task_id=None, status=pending, level=normal, dispatched_at=None, completed_at=None.""" self._create_chain(chain_mod) tasks_json = json.dumps([{"team": "dev2-team", "desc": "정규화 확인"}]) args = argparse.Namespace(chain="base-chain", name="Phase A", tasks=tasks_json) chain_mod.cmd_add_phase(args) chain_file = tmp_path / "memory" / "chains" / "base-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) task = data["phases"][0]["tasks"][0] assert task["task_id"] is None assert task["status"] == "pending" assert task["level"] == "normal" assert task["dispatched_at"] is None assert task["completed_at"] is None def test_tasks_level_preserved(self, chain_mod, tmp_path): """level=critical 지정 시 정규화 후에도 critical이 보존된다.""" self._create_chain(chain_mod) tasks_json = json.dumps([{"team": "dev1-team", "desc": "중요 작업", "level": "critical"}]) args = argparse.Namespace(chain="base-chain", name="Critical Phase", tasks=tasks_json) chain_mod.cmd_add_phase(args) chain_file = tmp_path / "memory" / "chains" / "base-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) task = data["phases"][0]["tasks"][0] assert task["level"] == "critical" def test_invalid_json_exits(self, chain_mod, tmp_path): """tasks JSON 파싱 실패 시 sys.exit(1)이 발생한다.""" self._create_chain(chain_mod) args = argparse.Namespace(chain="base-chain", name="Bad Phase", tasks="{not valid json}") with pytest.raises(SystemExit) as exc_info: chain_mod.cmd_add_phase(args) assert exc_info.value.code == 1 def test_missing_team_field_exits(self, chain_mod, tmp_path): """task 항목에 team 필드가 누락된 경우 sys.exit(1)이 발생한다.""" self._create_chain(chain_mod) tasks_json = json.dumps([{"desc": "팀 없음"}]) args = argparse.Namespace(chain="base-chain", name="No Team Phase", tasks=tasks_json) with pytest.raises(SystemExit) as exc_info: chain_mod.cmd_add_phase(args) assert exc_info.value.code == 1 def test_missing_desc_field_exits(self, chain_mod, tmp_path): """task 항목에 desc 필드가 누락된 경우 sys.exit(1)이 발생한다.""" self._create_chain(chain_mod) tasks_json = json.dumps([{"team": "dev1-team"}]) args = argparse.Namespace(chain="base-chain", name="No Desc Phase", tasks=tasks_json) with pytest.raises(SystemExit) as exc_info: chain_mod.cmd_add_phase(args) assert exc_info.value.code == 1 def test_chain_not_found_exits(self, chain_mod, tmp_path): """존재하지 않는 체인에 Phase 추가 시 sys.exit(1)이 발생한다.""" tasks_json = json.dumps([{"team": "dev1-team", "desc": "작업"}]) args = argparse.Namespace(chain="nonexistent-chain", name="Phase X", tasks=tasks_json) with pytest.raises(SystemExit) as exc_info: chain_mod.cmd_add_phase(args) assert exc_info.value.code == 1 # --------------------------------------------------------------------------- # 3. TestTaskDone: 완료 마킹, 미완료 대기, Phase 전환, paused 무시 # --------------------------------------------------------------------------- class TestTaskDone: """cmd_task_done() 서브커맨드 테스트""" def _make_subprocess_mock(self, returncode=0, stdout=None, stderr=""): """subprocess.run mock 결과를 생성하는 헬퍼.""" mock_result = MagicMock() mock_result.returncode = returncode mock_result.stdout = stdout if stdout is not None else json.dumps({"task_id": "task-3.1"}) mock_result.stderr = stderr return mock_result def test_task_done_marks_completed(self, chain_mod, tmp_path): """task-done 호출 시 해당 task의 status=completed, completed_at이 설정된다.""" _setup_chain_with_phases(tmp_path) chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" args = argparse.Namespace(chain="test-chain", task="task-1.1") chain_mod.cmd_task_done(args) chain_file = tmp_path / "memory" / "chains" / "test-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) task = data["phases"][0]["tasks"][0] assert task["status"] == "completed" assert task["completed_at"] is not None def test_not_all_complete_waits(self, chain_mod, tmp_path, capsys): """일부 task만 완료 시 '남음' 메시지가 출력되고 Phase 전환이 일어나지 않는다.""" _setup_chain_with_phases(tmp_path) chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" args = argparse.Namespace(chain="test-chain", task="task-1.1") chain_mod.cmd_task_done(args) captured = capsys.readouterr() assert "남음" in captured.out chain_file = tmp_path / "memory" / "chains" / "test-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["current_phase_idx"] == 0 assert data["phases"][0]["status"] == "in_progress" def test_all_tasks_done_transitions_phase(self, chain_mod, tmp_path, capsys): """모든 task 완료 시 Phase[0] completed, Phase[1] in_progress, current_phase_idx=1로 전환된다.""" _setup_chain_with_phases(tmp_path) chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" mock_result = self._make_subprocess_mock(0, json.dumps({"task_id": "task-3.1"})) with patch.object(chain_mod, "subprocess") as mock_sub: mock_sub.run.return_value = mock_result # task-1.1 완료 (아직 task-2.1 남음) args1 = argparse.Namespace(chain="test-chain", task="task-1.1") chain_mod.cmd_task_done(args1) # task-2.1 완료 → 전팀 완료, Phase 전환 트리거 args2 = argparse.Namespace(chain="test-chain", task="task-2.1") chain_mod.cmd_task_done(args2) chain_file = tmp_path / "memory" / "chains" / "test-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["phases"][0]["status"] == "completed" assert data["phases"][1]["status"] == "in_progress" assert data["current_phase_idx"] == 1 def test_last_phase_done_completes_chain(self, chain_mod, tmp_path, capsys): """마지막 Phase의 모든 task 완료 시 chain status=completed, completed_at이 설정된다.""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) # 단일 Phase 체인 (Phase 1개, task 1개) data = { "chain_id": "single-phase-chain", "description": "단일 Phase 체인", "status": "active", "current_phase_idx": 0, "created_at": datetime.now().isoformat(), "phases": [ { "name": "Phase 1", "status": "in_progress", "tasks": [ { "team": "dev1-team", "task_id": "task-1.1", "description": "유일한 작업", "level": "normal", "status": "in_progress", "dispatched_at": "2026-01-01", "completed_at": None, } ], } ], } chain_file = chains_dir / "single-phase-chain.json" chain_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") chain_mod.CHAINS_DIR = chains_dir mock_result = self._make_subprocess_mock(0, json.dumps({"task_id": "task-3.1"})) with patch.object(chain_mod, "subprocess") as mock_sub: mock_sub.run.return_value = mock_result args = argparse.Namespace(chain="single-phase-chain", task="task-1.1") chain_mod.cmd_task_done(args) updated = json.loads(chain_file.read_text(encoding="utf-8")) assert updated["status"] == "completed" assert updated["completed_at"] is not None def test_paused_chain_ignores_task_done(self, chain_mod, tmp_path, capsys): """paused 상태의 체인은 task-done을 무시하고 stderr에 'paused' 메시지를 출력한다.""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) data = { "chain_id": "paused-chain", "description": "일시정지 체인", "status": "paused", "current_phase_idx": 0, "created_at": datetime.now().isoformat(), "phases": [ { "name": "Phase 1", "status": "in_progress", "tasks": [ { "team": "dev1-team", "task_id": "task-1.1", "description": "작업", "level": "normal", "status": "in_progress", "dispatched_at": "2026-01-01", "completed_at": None, } ], } ], } chain_file = chains_dir / "paused-chain.json" chain_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") chain_mod.CHAINS_DIR = chains_dir args = argparse.Namespace(chain="paused-chain", task="task-1.1") chain_mod.cmd_task_done(args) captured = capsys.readouterr() assert "paused" in captured.err # task 상태가 변경되지 않아야 함 updated = json.loads(chain_file.read_text(encoding="utf-8")) assert updated["phases"][0]["tasks"][0]["status"] == "in_progress" def test_task_not_found_exits(self, chain_mod, tmp_path): """존재하지 않는 task_id 완료 시 sys.exit(1)이 발생한다.""" _setup_chain_with_phases(tmp_path) chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" args = argparse.Namespace(chain="test-chain", task="task-99.9") with pytest.raises(SystemExit) as exc_info: chain_mod.cmd_task_done(args) assert exc_info.value.code == 1 def test_chain_not_found_exits(self, chain_mod, tmp_path): """존재하지 않는 체인에 task-done 호출 시 sys.exit(1)이 발생한다.""" chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" args = argparse.Namespace(chain="ghost-chain", task="task-1.1") with pytest.raises(SystemExit) as exc_info: chain_mod.cmd_task_done(args) assert exc_info.value.code == 1 def test_dispatch_error_pauses_chain(self, chain_mod, tmp_path, capsys): """Phase 전환 시 dispatch 실패가 발생하면 chain status=paused, error 필드가 설정된다.""" _setup_chain_with_phases(tmp_path) chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" mock_error_result = self._make_subprocess_mock(1, "", "error message") with patch.object(chain_mod, "subprocess") as mock_sub: mock_sub.run.return_value = mock_error_result # task-1.1 완료 args1 = argparse.Namespace(chain="test-chain", task="task-1.1") chain_mod.cmd_task_done(args1) # task-2.1 완료 → Phase 전환 시도 → dispatch 실패 args2 = argparse.Namespace(chain="test-chain", task="task-2.1") chain_mod.cmd_task_done(args2) chain_file = tmp_path / "memory" / "chains" / "test-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["status"] == "paused" assert "error" in data # --------------------------------------------------------------------------- # 4. TestStatus: 정상 출력 # --------------------------------------------------------------------------- class TestStatus: """cmd_status() 서브커맨드 테스트""" def test_status_output(self, chain_mod, tmp_path, capsys): """cmd_status 실행 시 JSON 출력에 chain_id와 status 필드가 포함된다.""" create_args = argparse.Namespace(id="status-chain", desc="상태 확인 테스트") chain_mod.cmd_create(create_args) # cmd_create의 [OK] 출력을 미리 소비하여 cmd_status 출력만 캡처한다 capsys.readouterr() status_args = argparse.Namespace(chain="status-chain") chain_mod.cmd_status(status_args) captured = capsys.readouterr() output = json.loads(captured.out) assert "chain_id" in output assert output["chain_id"] == "status-chain" assert "status" in output def test_status_nonexistent_exits(self, chain_mod, tmp_path): """존재하지 않는 체인의 status 조회 시 sys.exit(1)이 발생한다.""" args = argparse.Namespace(chain="ghost-chain") with pytest.raises(SystemExit) as exc_info: chain_mod.cmd_status(args) assert exc_info.value.code == 1 # --------------------------------------------------------------------------- # 5. TestList: 빈 목록, 여러 체인 # --------------------------------------------------------------------------- class TestList: """cmd_list() 서브커맨드 테스트""" def test_list_empty(self, chain_mod, tmp_path, capsys): """체인이 없을 때 '활성 체인이 없습니다' 메시지가 출력된다.""" args = argparse.Namespace() chain_mod.cmd_list(args) captured = capsys.readouterr() assert "활성 체인이 없습니다" in captured.out def test_list_multiple_chains(self, chain_mod, tmp_path, capsys): """여러 체인 생성 후 목록에 모두 포함되는지 확인한다.""" for i in range(3): create_args = argparse.Namespace(id=f"chain-{i}", desc=f"체인 {i}") chain_mod.cmd_create(create_args) # cmd_create의 [OK] 출력들을 미리 소비한다 capsys.readouterr() list_args = argparse.Namespace() chain_mod.cmd_list(list_args) captured = capsys.readouterr() output = json.loads(captured.out) chain_ids = [item["chain_id"] for item in output] assert "chain-0" in chain_ids assert "chain-1" in chain_ids assert "chain-2" in chain_ids def test_list_shows_correct_fields(self, chain_mod, tmp_path, capsys): """목록의 각 항목에 chain_id, status, current_phase_idx, total_phases 필드가 있다.""" create_args = argparse.Namespace(id="field-chain", desc="필드 확인") chain_mod.cmd_create(create_args) # cmd_create의 [OK] 출력을 미리 소비한다 capsys.readouterr() list_args = argparse.Namespace() chain_mod.cmd_list(list_args) captured = capsys.readouterr() output = json.loads(captured.out) assert len(output) == 1 item = output[0] assert "chain_id" in item assert "status" in item assert "current_phase_idx" in item assert "total_phases" in item # --------------------------------------------------------------------------- # 6. TestUpdateChainTask: dispatch.py _update_chain_task 함수 동작 # --------------------------------------------------------------------------- class TestUpdateChainTask: """dispatch.py의 _update_chain_task() 함수 테스트""" def _setup_chain_for_update(self, tmp_path, chain_id="update-chain"): """_update_chain_task 테스트용 체인 파일을 생성한다.""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) data = { "chain_id": chain_id, "description": "업데이트 테스트", "status": "active", "current_phase_idx": 0, "created_at": datetime.now().isoformat(), "phases": [ { "name": "Phase 1", "status": "in_progress", "tasks": [ { "team": "dev1-team", "task_id": None, "description": "dispatch 대기 작업", "level": "normal", "status": "pending", "dispatched_at": None, "completed_at": None, } ], } ], } chain_file = chains_dir / f"{chain_id}.json" chain_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") return data def test_update_chain_task_sets_task_id(self, dispatch_mod, tmp_path): """pending task에 task_id, status=in_progress, dispatched_at이 설정된다.""" self._setup_chain_for_update(tmp_path) dispatch_mod._update_chain_task("update-chain", "dev1-team", "task-5.1") chain_file = tmp_path / "memory" / "chains" / "update-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) task = data["phases"][0]["tasks"][0] assert task["task_id"] == "task-5.1" assert task["status"] == "in_progress" assert task["dispatched_at"] is not None def test_update_chain_task_chain_not_found(self, dispatch_mod, tmp_path): """존재하지 않는 체인 파일일 때 에러 없이 경고만 출력하고 정상 종료된다.""" # 체인 파일을 생성하지 않고 호출 — 예외 없이 통과해야 함 dispatch_mod._update_chain_task("nonexistent-chain", "dev1-team", "task-1.1") def test_update_chain_task_does_not_overwrite_existing_task_id(self, dispatch_mod, tmp_path): """이미 task_id가 설정된 task는 _update_chain_task에 의해 덮어쓰여지지 않는다.""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) data = { "chain_id": "no-overwrite-chain", "description": "덮어쓰기 방지 테스트", "status": "active", "current_phase_idx": 0, "created_at": datetime.now().isoformat(), "phases": [ { "name": "Phase 1", "status": "in_progress", "tasks": [ { "team": "dev1-team", "task_id": "task-already-set", "description": "이미 dispatch된 작업", "level": "normal", "status": "in_progress", "dispatched_at": "2026-01-01T00:00:00", "completed_at": None, } ], } ], } chain_file = chains_dir / "no-overwrite-chain.json" chain_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") # task_id가 이미 있는 task에는 업데이트가 적용되지 않아야 함 dispatch_mod._update_chain_task("no-overwrite-chain", "dev1-team", "task-new-id") updated = json.loads(chain_file.read_text(encoding="utf-8")) task = updated["phases"][0]["tasks"][0] assert task["task_id"] == "task-already-set" def test_update_chain_task_correct_team_match(self, dispatch_mod, tmp_path): """여러 팀의 task 중 지정된 team의 task만 업데이트된다.""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) data = { "chain_id": "multi-team-chain", "description": "다팀 테스트", "status": "active", "current_phase_idx": 0, "created_at": datetime.now().isoformat(), "phases": [ { "name": "Phase 1", "status": "in_progress", "tasks": [ { "team": "dev1-team", "task_id": None, "description": "dev1 작업", "level": "normal", "status": "pending", "dispatched_at": None, "completed_at": None, }, { "team": "dev2-team", "task_id": None, "description": "dev2 작업", "level": "normal", "status": "pending", "dispatched_at": None, "completed_at": None, }, ], } ], } chain_file = chains_dir / "multi-team-chain.json" chain_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") dispatch_mod._update_chain_task("multi-team-chain", "dev1-team", "task-dev1.1") updated = json.loads(chain_file.read_text(encoding="utf-8")) tasks = updated["phases"][0]["tasks"] dev1_task = next(t for t in tasks if t["team"] == "dev1-team") dev2_task = next(t for t in tasks if t["team"] == "dev2-team") assert dev1_task["task_id"] == "task-dev1.1" assert dev1_task["status"] == "in_progress" # dev2 task는 변경되지 않아야 함 assert dev2_task["task_id"] is None assert dev2_task["status"] == "pending" # --------------------------------------------------------------------------- # 7. TestDispatchPhaseTaskFile: _dispatch_phase의 --task-file 사용 테스트 # --------------------------------------------------------------------------- class TestDispatchPhaseTaskFile: """_dispatch_phase()가 --task-file 방식을 사용하는지 테스트""" def _setup_chain_with_pending_phase(self, tmp_path, chain_id="tf-chain", description="작업 설명"): """dispatch 대상 Phase가 있는 체인 파일을 생성하는 헬퍼.""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) data = { "chain_id": chain_id, "description": "태스크 파일 테스트", "status": "active", "current_phase_idx": 0, "created_at": datetime.now().isoformat(), "phases": [ { "name": "Phase 1", "status": "pending", "tasks": [ { "team": "dev1-team", "task_id": None, "description": description, "level": "normal", "status": "pending", "dispatched_at": None, "completed_at": None, } ], } ], } chain_file = chains_dir / f"{chain_id}.json" chain_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") return data def test_dispatch_phase_writes_task_file(self, chain_mod, tmp_path): """_dispatch_phase 호출 시 description이 memory/tasks/ 하위 파일로 저장된다.""" description = "파일로 저장될 작업 설명" self._setup_chain_with_pending_phase(tmp_path, description=description) chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" chain_mod.WORKSPACE = tmp_path chain_file = tmp_path / "memory" / "chains" / "tf-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"task_id": "task-9.1"}) mock_result.stderr = "" with patch.object(chain_mod, "subprocess") as mock_sub: mock_sub.run.return_value = mock_result chain_mod._dispatch_phase(data, 0, "tf-chain") # memory/tasks/ 디렉토리 안에 파일이 생성되었는지 확인 tasks_dir = tmp_path / "memory" / "tasks" task_files = list(tasks_dir.glob("chain-tf-chain-phase0-dev1-team.md")) assert len(task_files) == 1, f"태스크 파일이 생성되지 않았습니다. tasks_dir 내용: {list(tasks_dir.iterdir())}" # 파일 내용이 description과 일치하는지 확인 content = task_files[0].read_text(encoding="utf-8") assert content == description def test_dispatch_phase_uses_task_file_flag(self, chain_mod, tmp_path): """_dispatch_phase 호출 시 subprocess.run에 --task-file 플래그가 사용된다.""" self._setup_chain_with_pending_phase(tmp_path, description="task-file 플래그 확인용 작업") chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" chain_mod.WORKSPACE = tmp_path chain_file = tmp_path / "memory" / "chains" / "tf-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"task_id": "task-9.2"}) mock_result.stderr = "" with patch.object(chain_mod, "subprocess") as mock_sub: mock_sub.run.return_value = mock_result chain_mod._dispatch_phase(data, 0, "tf-chain") assert mock_sub.run.called, "subprocess.run이 호출되지 않았습니다." call_args = mock_sub.run.call_args # subprocess 리스트 방식: 첫 번째 인자는 ["python3", ..., "--task-file", ...] cmd_list = call_args[0][0] assert "--task-file" in cmd_list, f"--task-file 플래그가 cmd에 없습니다: {cmd_list}" assert ( "--task" not in cmd_list or cmd_list[cmd_list.index("--task-file") - 1] != "--task" ), f"구식 --task 플래그가 여전히 cmd에 있습니다: {cmd_list}" def test_dispatch_phase_special_chars(self, chain_mod, tmp_path): """특수문자(', \", $, 백틱)가 포함된 description이 파일을 통해 깨지지 않고 저장된다.""" special_description = """특수문자 테스트: 작은따옴표('), 큰따옴표("), 달러($VAR), 백틱(`cmd`), 줄바꿈 두 번째 줄도 포함""" self._setup_chain_with_pending_phase( tmp_path, chain_id="tf-chain", description=special_description, ) chain_mod.CHAINS_DIR = tmp_path / "memory" / "chains" chain_mod.WORKSPACE = tmp_path chain_file = tmp_path / "memory" / "chains" / "tf-chain.json" data = json.loads(chain_file.read_text(encoding="utf-8")) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"task_id": "task-9.3"}) mock_result.stderr = "" with patch.object(chain_mod, "subprocess") as mock_sub: mock_sub.run.return_value = mock_result chain_mod._dispatch_phase(data, 0, "tf-chain") tasks_dir = tmp_path / "memory" / "tasks" task_files = list(tasks_dir.glob("chain-tf-chain-phase0-dev1-team.md")) assert len(task_files) == 1, "특수문자 description에 대한 태스크 파일이 생성되지 않았습니다." content = task_files[0].read_text(encoding="utf-8") assert ( content == special_description ), f"파일 내용이 원본 description과 다릅니다.\n예상: {special_description!r}\n실제: {content!r}" # --------------------------------------------------------------------------- # 8. TestCronNotifyGracefulSkip: _cron_notify() ANU_KEY 없으면 graceful skip # --------------------------------------------------------------------------- class TestCronNotifyGracefulSkip: """_cron_notify() ANU_KEY 없으면 graceful skip (task-448.1)""" def test_cron_notify_skips_when_no_anu_key(self, chain_mod, tmp_path): """ANU_KEY가 비어있으면 _cron_notify가 subprocess 호출 없이 리턴""" chain_mod.ANU_KEY = "" with patch.object(chain_mod, "subprocess") as mock_sub: chain_mod._cron_notify("테스트 알림") mock_sub.run.assert_not_called() def test_cron_notify_works_when_anu_key_set(self, chain_mod, tmp_path): """ANU_KEY가 설정되어 있으면 _cron_notify가 subprocess 정상 호출""" chain_mod.ANU_KEY = "test-key-123" mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" mock_result.stderr = "" with patch.object(chain_mod, "subprocess") as mock_sub: mock_sub.run.return_value = mock_result mock_sub.TimeoutExpired = subprocess.TimeoutExpired chain_mod._cron_notify("테스트 알림") mock_sub.run.assert_called_once() # --------------------------------------------------------------------------- # 9. TestShellInjectionDefense: _dispatch_phase 쉘 인젝션 방어 # --------------------------------------------------------------------------- class TestShellInjectionDefense: """_dispatch_phase 쉘 인젝션 방어 (task-448.1)""" def test_dispatch_phase_rejects_injection_in_chain_id(self, chain_mod, tmp_path): """chain_id에 '; rm -rf /' 같은 값 넣으면 ValueError 발생""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) chain_mod.CHAINS_DIR = chains_dir chain_mod.WORKSPACE = tmp_path data = { "phases": [ { "name": "Phase 1", "tasks": [ { "team": "dev1-team", "description": "test", "level": "normal", "status": "pending", } ], } ] } with pytest.raises(ValueError): chain_mod._dispatch_phase(data, 0, '"; rm -rf /') def test_dispatch_phase_rejects_injection_in_team(self, chain_mod, tmp_path): """team에 인젝션 문자열을 넣으면 ValueError 발생""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) chain_mod.CHAINS_DIR = chains_dir chain_mod.WORKSPACE = tmp_path data = { "phases": [ { "name": "Phase 1", "tasks": [ { "team": "dev1-team; echo hacked", "description": "test", "level": "normal", "status": "pending", } ], } ] } with pytest.raises(ValueError): chain_mod._dispatch_phase(data, 0, "safe-chain-id") def test_dispatch_phase_accepts_valid_ids(self, chain_mod, tmp_path): """정상적인 chain_id, team, level은 통과""" chains_dir = tmp_path / "memory" / "chains" chains_dir.mkdir(parents=True, exist_ok=True) (tmp_path / "memory" / "tasks").mkdir(parents=True, exist_ok=True) chain_mod.CHAINS_DIR = chains_dir chain_mod.WORKSPACE = tmp_path data = { "phases": [ { "name": "Phase 1", "tasks": [ { "team": "dev1-team", "description": "valid test", "level": "normal", "status": "pending", } ], } ] } mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"task_id": "task-1.1"}) mock_result.stderr = "" with patch.object(chain_mod, "subprocess") as mock_sub: mock_sub.run.return_value = mock_result # ValueError 없이 정상 실행 results = chain_mod._dispatch_phase(data, 0, "valid-chain-123") assert len(results) == 1 assert results[0]["status"] == "dispatched"