""" test_chain_manager.py chain_manager.py 순차 작업 체이닝 관리 유틸리티 단위 테스트 테스트 항목: - TestCreate: 정상 생성, max_tasks 초과 거부, 중복 chain_id 거부 - TestNext: 다음 pending 반환, QC FAIL 시 stalled, chain_complete 반환, no_chain 반환, 동일 task_file 중복 차단 - TestUpdate: running/done/failed/stalled 상태 변경 - TestCheckStalled: 정체 작업 검출, 정체 없음 - TestList: 체인 목록 출력 - TestLock: 동시 접근 시 순차 처리 - TestBackup: .bak 파일 생성 확인 """ import argparse import json import os import sys import types from datetime import datetime, timedelta from pathlib import Path from unittest.mock import MagicMock, patch import pytest # --------------------------------------------------------------------------- # 헬퍼: chain_manager 모듈을 격리된 WORKSPACE로 로드 # --------------------------------------------------------------------------- def _load_chain_manager(tmp_path: Path) -> types.ModuleType: """chain_manager 모듈을 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)) # 매 테스트마다 깨끗한 모듈 상태를 보장 for mod_name in list(sys.modules.keys()): if mod_name == "chain_manager": del sys.modules[mod_name] # ANU_KEY는 모듈 레벨에서 크래시하지 않지만, # create/next 등 cokacdir 호출 함수에서 필요하므로 더미 키를 주입한다. os.environ.setdefault("COKACDIR_KEY_ANU", "test-dummy-key") import chain_manager as cm cm.WORKSPACE = tmp_path cm.CHAINS_DIR = tmp_path / "memory" / "chains" cm.CHAINS_DIR.mkdir(parents=True, exist_ok=True) return cm # --------------------------------------------------------------------------- # fixture # --------------------------------------------------------------------------- @pytest.fixture() def cm(tmp_path): """격리된 WORKSPACE를 사용하는 chain_manager 모듈을 반환한다.""" return _load_chain_manager(tmp_path) @pytest.fixture() def chains_dir(tmp_path): """체인 파일 저장 디렉토리.""" d = tmp_path / "memory" / "chains" d.mkdir(parents=True, exist_ok=True) return d # --------------------------------------------------------------------------- # 헬퍼 함수들 # --------------------------------------------------------------------------- def _make_chain_data( chain_id: str = "test-chain", tasks: list | None = None, status: str = "active", max_tasks: int = 10, scope: str = "테스트 범위", watchdog_cron_id: str | None = "cron-001", ) -> dict: """테스트용 체인 데이터를 생성한다.""" if tasks is None: tasks = [ { "order": 1, "task_file": "memory/tasks/dispatch-001.md", "team": "dev1-team", "status": "pending", "task_id": None, "gate": "auto", "started_at": None, "completed_at": None, } ] data: dict = { "chain_id": chain_id, "created_by": "anu", "created_at": datetime.now().isoformat(), "status": status, "scope": scope, "max_tasks": max_tasks, "tasks": tasks, } if watchdog_cron_id is not None: data["watchdog_cron_id"] = watchdog_cron_id return data def _write_chain(chains_dir: Path, data: dict) -> Path: """체인 데이터를 파일로 저장하고 경로를 반환한다.""" chain_id = data["chain_id"] # chain- 형식이면 그대로, 아니면 chain- 접두사 제거 if chain_id.startswith("chain-"): filename = f"{chain_id}.json" else: filename = f"chain-{chain_id}.json" path = chains_dir / filename path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return path # --------------------------------------------------------------------------- # 1. TestCreate # --------------------------------------------------------------------------- class TestCreate: """cmd_create() 서브커맨드 테스트""" def test_create_normal(self, cm, tmp_path, chains_dir): """정상 생성: 체인 파일이 올바른 스키마로 생성된다.""" tasks = [ {"order": 1, "task_file": "memory/tasks/t1.md", "team": "dev1-team", "gate": "auto"}, {"order": 2, "task_file": "memory/tasks/t2.md", "team": "dev2-team", "gate": "none"}, ] args = argparse.Namespace( chain_id="chain-20260307-001", tasks=json.dumps(tasks), scope="테스트 범위", created_by="anu", max_tasks=10, ) with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = '{"cron_id": "cron-abc123"}' mock_sub.run.return_value = mock_result cm.cmd_create(args) chain_file = chains_dir / "chain-chain-20260307-001.json" assert chain_file.exists() data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["chain_id"] == "chain-20260307-001" assert data["status"] == "active" assert data["scope"] == "테스트 범위" assert data["created_by"] == "anu" assert data["max_tasks"] == 10 assert len(data["tasks"]) == 2 # task 스키마 정규화 확인 t = data["tasks"][0] assert t["status"] == "pending" assert t["task_id"] is None assert t["started_at"] is None assert t["completed_at"] is None def test_create_registers_watchdog_cron(self, cm, tmp_path, chains_dir): """create 성공 시 cokacdir watchdog cron이 등록되고 cron_id가 체인 파일에 저장된다.""" tasks = [{"order": 1, "task_file": "memory/tasks/t1.md", "team": "dev1-team", "gate": "auto"}] args = argparse.Namespace( chain_id="chain-cron-test", tasks=json.dumps(tasks), scope="cron 테스트", created_by="anu", max_tasks=10, ) with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = json.dumps({"cron_id": "cron-watchdog-001"}) mock_sub.run.return_value = mock_result cm.cmd_create(args) # subprocess.run이 호출되었는지 확인 assert mock_sub.run.called chain_file = chains_dir / "chain-chain-cron-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) assert "watchdog_cron_id" in data def test_create_max_tasks_exceeded(self, cm, tmp_path): """max_tasks 초과 시 exit 1이 발생한다.""" tasks = [ {"order": i + 1, "task_file": f"memory/tasks/t{i}.md", "team": "dev1-team", "gate": "auto"} for i in range(5) ] args = argparse.Namespace( chain_id="chain-overflow", tasks=json.dumps(tasks), scope="초과 테스트", created_by="anu", max_tasks=3, # tasks 5개지만 max_tasks=3 ) with pytest.raises(SystemExit) as exc_info: cm.cmd_create(args) assert exc_info.value.code == 1 def test_create_duplicate_chain_id(self, cm, tmp_path, chains_dir): """동일 chain_id 중복 생성 시 exit 1이 발생한다.""" tasks = [{"order": 1, "task_file": "memory/tasks/t1.md", "team": "dev1-team", "gate": "auto"}] args = argparse.Namespace( chain_id="chain-dup", tasks=json.dumps(tasks), scope="중복 테스트", created_by="anu", max_tasks=10, ) with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = '{"cron_id": "cron-001"}' mock_sub.run.return_value = mock_result cm.cmd_create(args) # 첫 번째 생성 # 두 번째 생성 시 중복 오류 with pytest.raises(SystemExit) as exc_info: cm.cmd_create(args) assert exc_info.value.code == 1 def test_create_invalid_tasks_json(self, cm, tmp_path): """잘못된 tasks JSON 시 exit 1이 발생한다.""" args = argparse.Namespace( chain_id="chain-bad-json", tasks="{not valid json}", scope="JSON 오류 테스트", created_by="anu", max_tasks=10, ) with pytest.raises(SystemExit) as exc_info: cm.cmd_create(args) assert exc_info.value.code == 1 # --------------------------------------------------------------------------- # 2. TestNext # --------------------------------------------------------------------------- class TestNext: """cmd_next() 서브커맨드 테스트""" def _setup_chain( self, chains_dir: Path, chain_id: str = "chain-test", tasks: list | None = None, chain_status: str = "active", ) -> Path: """테스트용 체인 파일을 생성하는 헬퍼.""" if tasks is None: tasks = [ { "order": 1, "task_file": "memory/tasks/task-001.md", "team": "dev1-team", "status": "running", "task_id": "task-10.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, }, { "order": 2, "task_file": "memory/tasks/task-002.md", "team": "dev2-team", "status": "pending", "task_id": None, "gate": "auto", "started_at": None, "completed_at": None, }, ] data = { "chain_id": chain_id, "created_by": "anu", "created_at": datetime.now().isoformat(), "status": chain_status, "scope": "테스트", "max_tasks": 10, "watchdog_cron_id": "cron-watch-001", "tasks": tasks, } path = chains_dir / f"{chain_id}.json" path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return path def test_next_returns_dispatch(self, cm, tmp_path, chains_dir, capsys): """다음 pending task가 있으면 action=dispatch를 반환한다.""" self._setup_chain(chains_dir) args = argparse.Namespace(task_id="task-10.1") # 보고서 파일 (FAIL 없음) reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-10.1.md").write_text("작업 완료. 성공.", encoding="utf-8") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" assert output["task_file"] == "memory/tasks/task-002.md" assert output["team"] == "dev2-team" assert "chain_id" in output def test_next_qc_fail_triggers_retry(self, cm, tmp_path, chains_dir, capsys): """gate=auto 조건에서 보고서에 FAIL 키워드가 있으면 재위임(retry)한다 (F12).""" self._setup_chain(chains_dir) args = argparse.Namespace(task_id="task-10.1") # 보고서 파일에 FAIL 포함 reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-10.1.md").write_text("작업 완료. QC FAIL: 오류 발견.", encoding="utf-8") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" assert output["retry_attempt"] == 1 assert output["task_id"] == "task-10.1" # 체인 상태는 stalled가 아닌 active여야 함 chain_file = chains_dir / "chain-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["status"] == "active" # target_task의 retry_count가 1이어야 함 target_task = next(t for t in data["tasks"] if t.get("task_id") == "task-10.1") assert target_task["retry_count"] == 1 # target_task의 status가 "running"이어야 함 (다시 실행 중) assert target_task["status"] == "running" def test_next_qc_fail_out_of_scope_does_not_stall(self, cm, tmp_path, chains_dir, capsys): """gate=auto 조건에서 'QC FAIL (범위 외)'는 체인을 stall하지 않는다.""" self._setup_chain(chains_dir) args = argparse.Namespace(task_id="task-10.1") # 보고서 파일에 "QC FAIL (범위 외)" 포함 — stall되지 않아야 함 reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-10.1.md").write_text( "작업 완료. QC FAIL (범위 외 기존 테스트 1건 — 에스컬레이션).", encoding="utf-8" ) cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" # stalled가 아님 def test_next_gate_none_skips_qc(self, cm, tmp_path, chains_dir, capsys): """gate=none이면 보고서 내용에 관계없이 다음 pending task를 반환한다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-001.md", "team": "dev1-team", "status": "running", "task_id": "task-20.1", "gate": "none", # gate=none "started_at": datetime.now().isoformat(), "completed_at": None, }, { "order": 2, "task_file": "memory/tasks/task-002.md", "team": "dev2-team", "status": "pending", "task_id": None, "gate": "none", "started_at": None, "completed_at": None, }, ] self._setup_chain(chains_dir, tasks=tasks) args = argparse.Namespace(task_id="task-20.1") # 보고서에 FAIL이 있어도 gate=none이면 무시 reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-20.1.md").write_text("QC FAIL 가 있어도 무시해야 함", encoding="utf-8") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" def test_next_chain_complete(self, cm, tmp_path, chains_dir, capsys): """다음 pending task가 없으면 action=chain_complete를 반환하고 체인이 completed된다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-001.md", "team": "dev1-team", "status": "running", "task_id": "task-30.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, } # pending task 없음 ] self._setup_chain(chains_dir, tasks=tasks) args = argparse.Namespace(task_id="task-30.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-30.1.md").write_text("작업 완료.", encoding="utf-8") with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" mock_sub.run.return_value = mock_result cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "chain_complete" assert "chain_id" in output # 체인 상태 confirmed chain_file = chains_dir / "chain-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["status"] == "completed" def test_next_removes_watchdog_cron_on_complete(self, cm, tmp_path, chains_dir): """chain_complete 시 watchdog cron이 제거된다 (cokacdir --cron-remove 호출).""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-001.md", "team": "dev1-team", "status": "running", "task_id": "task-31.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, } ] self._setup_chain(chains_dir, tasks=tasks) args = argparse.Namespace(task_id="task-31.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-31.1.md").write_text("완료.", encoding="utf-8") with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" mock_sub.run.return_value = mock_result cm.cmd_next(args) # cokacdir --cron-remove가 호출되었는지 확인 called_cmds = [str(call) for call in mock_sub.run.call_args_list] assert any("cron-remove" in cmd for cmd in called_cmds) def test_next_no_chain(self, cm, tmp_path, chains_dir, capsys): """어떤 체인에도 없는 task_id이면 action=no_chain을 반환한다 (exit 0).""" args = argparse.Namespace(task_id="task-unknown-999") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "no_chain" assert output["task_id"] == "task-unknown-999" def test_next_duplicate_task_file_blocked(self, cm, tmp_path, chains_dir, capsys): """이미 running/done 상태인 task의 task_file과 동일한 task_file을 가진 pending이 차단된다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/same-file.md", "team": "dev1-team", "status": "running", "task_id": "task-40.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, }, { "order": 2, "task_file": "memory/tasks/same-file.md", # 동일한 task_file "team": "dev2-team", "status": "pending", "task_id": None, "gate": "auto", "started_at": None, "completed_at": None, }, ] self._setup_chain(chains_dir, tasks=tasks) args = argparse.Namespace(task_id="task-40.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-40.1.md").write_text("완료.", encoding="utf-8") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "stalled" assert "duplicate task_file" in output["reason"] def test_next_marks_completed_at(self, cm, tmp_path, chains_dir): """next 호출 시 완료된 task의 status=done, completed_at이 설정된다.""" self._setup_chain(chains_dir) args = argparse.Namespace(task_id="task-10.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-10.1.md").write_text("완료.", encoding="utf-8") cm.cmd_next(args) chain_file = chains_dir / "chain-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) done_task = next(t for t in data["tasks"] if t.get("task_id") == "task-10.1") assert done_task["status"] == "done" assert done_task["completed_at"] is not None def test_next_idempotency_already_done(self, cm, tmp_path, chains_dir, capsys): """이미 done인 task에 대해 next를 다시 호출하면 already_done을 반환한다.""" self._setup_chain(chains_dir) args = argparse.Namespace(task_id="task-10.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-10.1.md").write_text("완료.", encoding="utf-8") # 첫 번째 호출 cm.cmd_next(args) captured = capsys.readouterr() output1 = json.loads(captured.out) assert output1["action"] == "dispatch" # 두 번째 호출 (이미 done) cm.cmd_next(args) captured = capsys.readouterr() output2 = json.loads(captured.out) assert output2["action"] == "already_done" def test_next_idempotency_next_already_running(self, cm, tmp_path, chains_dir, capsys): """다음 task가 이미 running이면 dispatch하지 않는다.""" # 두 번째 task가 이미 running인 상태로 설정 tasks = [ { "order": 1, "task_file": "memory/tasks/task-10.1.md", "team": "dev1-team", "status": "running", "task_id": "task-10.1", "gate": "auto", }, { "order": 2, "task_file": "memory/tasks/task-10.2.md", "team": "dev1-team", "status": "running", # 이미 running "task_id": "task-10.2", "gate": "auto", }, ] self._setup_chain(chains_dir, tasks=tasks) args = argparse.Namespace(task_id="task-10.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-10.1.md").write_text("완료.", encoding="utf-8") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "already_running" def test_next_creates_missing_task_file_from_original(self, cm, tmp_path, chains_dir, capsys): """task_file이 없을 때 original_task_file에서 복사한다.""" # 원본 지시서 파일 생성 original_task_file = tmp_path / "memory" / "tasks" / "task-20.md" original_task_file.parent.mkdir(parents=True, exist_ok=True) original_task_file.write_text("# 원본 지시서\n\nPhase 1: ...\nPhase 2: ...", encoding="utf-8") # 체인 생성 (original_task_file 포함) tasks = [ { "order": 1, "task_file": "memory/tasks/task-20.1.md", "team": "dev1-team", "status": "running", "task_id": "task-20.1", "gate": "auto", "started_at": datetime.now().isoformat(), }, { "order": 2, "task_file": "memory/tasks/task-20.2.md", # 존재하지 않음 "team": "dev1-team", "status": "pending", "task_id": "task-20.2", "gate": "auto", }, ] chain_file = chains_dir / "chain-test.json" data = { "chain_id": "chain-test", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "", "max_tasks": 10, "original_task_file": "memory/tasks/task-20.md", "tasks": tasks, } chain_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") # Phase 1 완료 처리 args = argparse.Namespace(task_id="task-20.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-20.1.md").write_text("완료.", encoding="utf-8") cm.cmd_next(args) # Phase 2 지시서 파일이 자동 생성되었는지 확인 phase2_task_file = tmp_path / "memory" / "tasks" / "task-20.2.md" assert phase2_task_file.exists(), "Phase 2 지시서 파일이 자동 생성되어야 함" # 내용이 원본과 동일한지 확인 assert phase2_task_file.read_text(encoding="utf-8") == original_task_file.read_text(encoding="utf-8") # dispatch 액션 반환 확인 captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" assert output["task_file"] == "memory/tasks/task-20.2.md" # --------------------------------------------------------------------------- # 3. TestUpdate # --------------------------------------------------------------------------- class TestUpdate: """cmd_update() 서브커맨드 테스트""" def _setup_chain_with_task(self, chains_dir: Path, task_id: str = "task-50.1") -> Path: """update 테스트용 체인 파일을 생성한다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-x.md", "team": "dev1-team", "status": "pending", "task_id": task_id, "gate": "auto", "started_at": None, "completed_at": None, } ] data = { "chain_id": "chain-update-test", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "업데이트 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-001", "tasks": tasks, } path = chains_dir / "chain-update-test.json" path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return path def test_update_to_running(self, cm, tmp_path, chains_dir): """running으로 상태 변경 시 started_at이 설정된다.""" self._setup_chain_with_task(chains_dir) args = argparse.Namespace(task_id="task-50.1", status="running") cm.cmd_update(args) chain_file = chains_dir / "chain-update-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) task = data["tasks"][0] assert task["status"] == "running" assert task["started_at"] is not None def test_update_to_done(self, cm, tmp_path, chains_dir): """done으로 상태 변경 시 completed_at이 설정된다.""" self._setup_chain_with_task(chains_dir) args = argparse.Namespace(task_id="task-50.1", status="done") cm.cmd_update(args) chain_file = chains_dir / "chain-update-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) task = data["tasks"][0] assert task["status"] == "done" assert task["completed_at"] is not None def test_update_to_failed(self, cm, tmp_path, chains_dir): """failed로 상태 변경 시 status=failed로 업데이트된다.""" self._setup_chain_with_task(chains_dir) args = argparse.Namespace(task_id="task-50.1", status="failed") cm.cmd_update(args) chain_file = chains_dir / "chain-update-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) task = data["tasks"][0] assert task["status"] == "failed" def test_update_to_stalled(self, cm, tmp_path, chains_dir): """stalled로 상태 변경 시 status=stalled로 업데이트된다.""" self._setup_chain_with_task(chains_dir) args = argparse.Namespace(task_id="task-50.1", status="stalled") cm.cmd_update(args) chain_file = chains_dir / "chain-update-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) task = data["tasks"][0] assert task["status"] == "stalled" def test_update_task_not_found_exits(self, cm, tmp_path, chains_dir): """존재하지 않는 task_id 업데이트 시 exit 1이 발생한다.""" self._setup_chain_with_task(chains_dir) args = argparse.Namespace(task_id="task-nonexistent", status="done") with pytest.raises(SystemExit) as exc_info: cm.cmd_update(args) assert exc_info.value.code == 1 # --------------------------------------------------------------------------- # 4. TestCheckStalled # --------------------------------------------------------------------------- class TestCheckStalled: """cmd_check_stalled() 서브커맨드 테스트""" def _setup_chain_with_running_task( self, chains_dir: Path, chain_id: str = "chain-stalled-test", started_hours_ago: float = 3.0, ) -> Path: """running 상태 task가 있는 체인 파일을 생성한다.""" started_at = (datetime.now() - timedelta(hours=started_hours_ago)).isoformat() tasks = [ { "order": 1, "task_file": "memory/tasks/running-task.md", "team": "dev2-team", "status": "running", "task_id": "task-60.1", "gate": "auto", "started_at": started_at, "completed_at": None, } ] data = { "chain_id": chain_id, "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "정체 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-001", "tasks": tasks, } path = chains_dir / f"{chain_id}.json" path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return path def test_check_stalled_detects_stalled(self, cm, tmp_path, chains_dir, capsys): """max_hours 초과한 running 작업이 검출된다.""" self._setup_chain_with_running_task(chains_dir, started_hours_ago=3.0) args = argparse.Namespace(max_hours=2) cm.cmd_check_stalled(args) captured = capsys.readouterr() output = json.loads(captured.out) assert isinstance(output, list) assert len(output) == 1 assert output[0]["chain_id"] == "chain-stalled-test" assert output[0]["task_id"] == "task-60.1" assert output[0]["team"] == "dev2-team" assert output[0]["hours_elapsed"] >= 2 def test_check_stalled_no_stalled(self, cm, tmp_path, chains_dir, capsys): """max_hours 미만인 running 작업은 검출되지 않는다.""" self._setup_chain_with_running_task(chains_dir, started_hours_ago=1.0) args = argparse.Namespace(max_hours=2) cm.cmd_check_stalled(args) captured = capsys.readouterr() output = json.loads(captured.out) assert isinstance(output, list) assert len(output) == 0 def test_check_stalled_empty_result(self, cm, tmp_path, chains_dir, capsys): """활성 체인이 없으면 빈 배열을 출력한다.""" args = argparse.Namespace(max_hours=2) cm.cmd_check_stalled(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output == [] def test_check_stalled_skips_completed_chains(self, cm, tmp_path, chains_dir, capsys): """completed/stalled 체인의 task는 정체 검사에서 제외된다.""" started_at = (datetime.now() - timedelta(hours=5)).isoformat() tasks = [ { "order": 1, "task_file": "memory/tasks/t1.md", "team": "dev1-team", "status": "running", "task_id": "task-70.1", "gate": "auto", "started_at": started_at, "completed_at": None, } ] data = { "chain_id": "chain-completed", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "completed", # completed 체인 "scope": "완료 체인", "max_tasks": 10, "tasks": tasks, } path = chains_dir / "chain-completed.json" path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") args = argparse.Namespace(max_hours=2) cm.cmd_check_stalled(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output == [] # --------------------------------------------------------------------------- # 5. TestList # --------------------------------------------------------------------------- class TestList: """cmd_list() 서브커맨드 테스트""" def test_list_empty(self, cm, tmp_path, chains_dir, capsys): """체인이 없을 때 빈 배열을 출력한다.""" args = argparse.Namespace() cm.cmd_list(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output == [] def test_list_multiple_chains(self, cm, tmp_path, chains_dir, capsys): """여러 체인이 있을 때 모두 목록에 포함된다.""" for i in range(3): data = _make_chain_data(chain_id=f"chain-{i:03d}", scope=f"범위 {i}") _write_chain(chains_dir, data) args = argparse.Namespace() cm.cmd_list(args) captured = capsys.readouterr() output = json.loads(captured.out) assert len(output) == 3 chain_ids = [item["chain_id"] for item in output] assert "chain-000" in chain_ids assert "chain-001" in chain_ids assert "chain-002" in chain_ids def test_list_shows_required_fields(self, cm, tmp_path, chains_dir, capsys): """목록의 각 항목에 chain_id, status, scope, tasks 수 필드가 있다.""" data = _make_chain_data(chain_id="chain-fields") _write_chain(chains_dir, data) args = argparse.Namespace() cm.cmd_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 "scope" in item assert "task_count" in item def test_list_task_count_correct(self, cm, tmp_path, chains_dir, capsys): """task_count가 실제 tasks 배열의 길이와 일치한다.""" tasks = [ { "order": i + 1, "task_file": f"memory/tasks/t{i}.md", "team": "dev1-team", "status": "pending", "task_id": None, "gate": "auto", "started_at": None, "completed_at": None, } for i in range(4) ] data = _make_chain_data(chain_id="chain-count", tasks=tasks) _write_chain(chains_dir, data) args = argparse.Namespace() cm.cmd_list(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output[0]["task_count"] == 4 # --------------------------------------------------------------------------- # 6. TestLock: 동시 접근 시 순차 처리 # --------------------------------------------------------------------------- class TestLock: """lock 파일을 이용한 원자적 상태 업데이트 테스트""" def test_lock_file_created_and_removed(self, cm, tmp_path, chains_dir): """상태 업데이트 시 lock 파일이 생성되고 완료 후 삭제된다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/lock-test.md", "team": "dev1-team", "status": "pending", "task_id": "task-80.1", "gate": "auto", "started_at": None, "completed_at": None, } ] data = { "chain_id": "chain-lock-test", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "락 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-001", "tasks": tasks, } chain_file = chains_dir / "chain-lock-test.json" chain_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") args = argparse.Namespace(task_id="task-80.1", status="running") cm.cmd_update(args) # 작업 완료 후 lock 파일이 삭제되어야 함 lock_file = chains_dir / "chain-lock-test.lock" assert not lock_file.exists() def test_sequential_updates_consistency(self, cm, tmp_path, chains_dir): """순차 업데이트 시 최종 상태가 정확하다.""" tasks = [ { "order": i + 1, "task_file": f"memory/tasks/t{i}.md", "team": "dev1-team", "status": "pending", "task_id": f"task-{90 + i}.1", "gate": "auto", "started_at": None, "completed_at": None, } for i in range(3) ] data = { "chain_id": "chain-seq-test", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "순차 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-001", "tasks": tasks, } chain_file = chains_dir / "chain-seq-test.json" chain_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") # 순차적으로 3개 task 업데이트 for i in range(3): args = argparse.Namespace(task_id=f"task-{90 + i}.1", status="done") cm.cmd_update(args) final_data = json.loads(chain_file.read_text(encoding="utf-8")) for task in final_data["tasks"]: assert task["status"] == "done" # --------------------------------------------------------------------------- # 7. TestBackup: .bak 파일 생성 확인 # --------------------------------------------------------------------------- class TestBackup: """체인 파일 .bak 백업 기능 테스트""" def test_backup_created_on_update(self, cm, tmp_path, chains_dir): """.bak 파일이 update 직전에 생성된다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/bak-test.md", "team": "dev1-team", "status": "pending", "task_id": "task-100.1", "gate": "auto", "started_at": None, "completed_at": None, } ] data = { "chain_id": "chain-bak-test", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "백업 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-001", "tasks": tasks, } chain_file = chains_dir / "chain-bak-test.json" chain_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") args = argparse.Namespace(task_id="task-100.1", status="running") cm.cmd_update(args) bak_file = chains_dir / "chain-bak-test.json.bak" assert bak_file.exists() def test_backup_contains_original_data(self, cm, tmp_path, chains_dir): """.bak 파일에 업데이트 전 원본 데이터가 저장된다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/bak-content-test.md", "team": "dev1-team", "status": "pending", "task_id": "task-101.1", "gate": "auto", "started_at": None, "completed_at": None, } ] data = { "chain_id": "chain-bak-content", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "백업 내용 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-001", "tasks": tasks, } chain_file = chains_dir / "chain-bak-content.json" original_text = json.dumps(data, ensure_ascii=False, indent=2) chain_file.write_text(original_text, encoding="utf-8") args = argparse.Namespace(task_id="task-101.1", status="running") cm.cmd_update(args) bak_file = chains_dir / "chain-bak-content.json.bak" bak_data = json.loads(bak_file.read_text(encoding="utf-8")) # 백업에는 업데이트 전 상태(pending)가 있어야 함 assert bak_data["tasks"][0]["status"] == "pending" def test_backup_created_on_next(self, cm, tmp_path, chains_dir): """cmd_next 호출 시에도 .bak 파일이 생성된다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/bak-next-test.md", "team": "dev1-team", "status": "running", "task_id": "task-102.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, } ] data = { "chain_id": "chain-bak-next", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "next 백업 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-001", "tasks": tasks, } chain_file = chains_dir / "chain-bak-next.json" chain_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-102.1.md").write_text("완료.", encoding="utf-8") with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "" mock_sub.run.return_value = mock_result args = argparse.Namespace(task_id="task-102.1") cm.cmd_next(args) bak_file = chains_dir / "chain-bak-next.json.bak" assert bak_file.exists() # --------------------------------------------------------------------------- # 8. TestCheck: check 서브커맨드 테스트 # --------------------------------------------------------------------------- class TestCheck: """cmd_check() 서브커맨드 테스트 (읽기 전용)""" def _setup_chain(self, chains_dir: Path, chain_id: str, tasks: list) -> Path: """테스트용 체인 파일을 생성한다.""" data = { "chain_id": chain_id, "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "check 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-001", "tasks": tasks, } path = chains_dir / f"chain-{chain_id}.json" path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return path def test_check_in_chain_not_last(self, cm, tmp_path, chains_dir, capsys): """체인 소속이고 마지막이 아닌 경우: in_chain=true, is_last=false, next_task_id 반환.""" tasks = [ { "order": 1, "task_file": "memory/tasks/dispatch-436.2.md", "team": "dev1-team", "status": "running", "task_id": "task-436.2", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, }, { "order": 2, "task_file": "memory/tasks/dispatch-436.3.md", "team": "dev2-team", "status": "pending", "task_id": "task-436.3", "gate": "auto", "started_at": None, "completed_at": None, }, ] self._setup_chain(chains_dir, "remotion-migration", tasks) args = argparse.Namespace(task_id="task-436.2") cm.cmd_check(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["in_chain"] is True assert output["is_last"] is False assert output["chain_id"] == "remotion-migration" assert output["next_task_id"] == "task-436.3" def test_check_in_chain_is_last(self, cm, tmp_path, chains_dir, capsys): """체인 소속이고 마지막인 경우: in_chain=true, is_last=true, next_task_id=null.""" tasks = [ { "order": 1, "task_file": "memory/tasks/dispatch-436.1.md", "team": "dev1-team", "status": "done", "task_id": "task-436.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": datetime.now().isoformat(), }, { "order": 2, "task_file": "memory/tasks/dispatch-436.2.md", "team": "dev2-team", "status": "running", "task_id": "task-436.2", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, }, ] self._setup_chain(chains_dir, "remotion-migration-last", tasks) args = argparse.Namespace(task_id="task-436.2") cm.cmd_check(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["in_chain"] is True assert output["is_last"] is True assert output["chain_id"] == "remotion-migration-last" assert output["next_task_id"] is None def test_check_not_in_chain(self, cm, tmp_path, chains_dir, capsys): """어떤 체인에도 없는 task_id: in_chain=false, chain_id=null, next_task_id=null.""" args = argparse.Namespace(task_id="task-nonexistent-9999") cm.cmd_check(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["in_chain"] is False assert output["is_last"] is False assert output["chain_id"] is None assert output["next_task_id"] is None def test_check_next_task_id_none_when_pending(self, cm, tmp_path, chains_dir, capsys): """다음 task가 pending이고 task_id=None인 경우: next_task_id=null 반환.""" tasks = [ { "order": 1, "task_file": "memory/tasks/dispatch-500.1.md", "team": "dev1-team", "status": "running", "task_id": "task-500.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, }, { "order": 2, "task_file": "memory/tasks/dispatch-500.2.md", "team": "dev2-team", "status": "pending", "task_id": None, # task_id 미할당 "gate": "auto", "started_at": None, "completed_at": None, }, ] self._setup_chain(chains_dir, "pending-no-id-chain", tasks) args = argparse.Namespace(task_id="task-500.1") cm.cmd_check(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["in_chain"] is True assert output["is_last"] is False assert output["chain_id"] == "pending-no-id-chain" assert output["next_task_id"] is None def test_create_preserves_task_id(self, cm, tmp_path, chains_dir): """create 시 tasks JSON의 task_id가 체인 파일에 보존된다.""" tasks = [ {"task_id": "task-100.1", "order": 1, "task_file": "memory/tasks/test.md", "team": "dev1-team"}, {"task_id": "task-100.2", "order": 2, "task_file": "memory/tasks/test2.md", "team": "dev2-team"}, ] args = argparse.Namespace( chain_id="preserve-id-test", tasks=json.dumps(tasks), scope="task_id 보존 테스트", created_by="anu", max_tasks=10, ) with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = '{"cron_id": "cron-preserve"}' mock_sub.run.return_value = mock_result cm.cmd_create(args) chain_file = chains_dir / "chain-preserve-id-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["tasks"][0]["task_id"] == "task-100.1" assert data["tasks"][1]["task_id"] == "task-100.2" def test_check_finds_chain_by_task_id(self, cm, tmp_path, chains_dir, capsys): """create로 생성한 체인에서 task_id로 check 매칭이 성공한다.""" tasks = [ {"task_id": "task-200.1", "order": 1, "task_file": "memory/tasks/t200-1.md", "team": "dev1-team"}, {"task_id": "task-200.2", "order": 2, "task_file": "memory/tasks/t200-2.md", "team": "dev2-team"}, ] args_create = argparse.Namespace( chain_id="check-find-test", tasks=json.dumps(tasks), scope="check 매칭 테스트", created_by="anu", max_tasks=10, ) with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = '{"cron_id": "cron-find"}' mock_sub.run.return_value = mock_result cm.cmd_create(args_create) capsys.readouterr() # create 출력 비우기 args_check = argparse.Namespace(task_id="task-200.1") cm.cmd_check(args_check) captured = capsys.readouterr() output = json.loads(captured.out) assert output["in_chain"] is True assert output["is_last"] is False assert output["chain_id"] == "check-find-test" assert output["next_task_id"] == "task-200.2" def test_next_includes_task_id_in_output(self, cm, tmp_path, chains_dir, capsys): """cmd_next의 dispatch action 출력에 task_id 필드가 포함된다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-566.1.md", "team": "dev1-team", "status": "running", "task_id": "task-566.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, }, { "order": 2, "task_file": "memory/tasks/task-566.2.md", "team": "dev1-team", "status": "pending", "task_id": "task-566.2", "gate": "auto", "started_at": None, "completed_at": None, }, ] self._setup_chain(chains_dir, "chain-566-test", tasks) args = argparse.Namespace(task_id="task-566.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-566.1.md").write_text("작업 완료.", encoding="utf-8") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" # task_id 필드가 출력에 포함되어야 함 assert "task_id" in output, "dispatch action 출력에 task_id 필드가 없음" assert output["task_id"] == "task-566.2" def test_next_task_id_none_when_pending_has_no_id(self, cm, tmp_path, chains_dir, capsys): """다음 pending task의 task_id가 None이면 출력의 task_id도 None이어야 한다.""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-567.1.md", "team": "dev1-team", "status": "running", "task_id": "task-567.1", "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, }, { "order": 2, "task_file": "memory/tasks/task-567.2.md", "team": "dev1-team", "status": "pending", "task_id": None, # task_id 미할당 "gate": "auto", "started_at": None, "completed_at": None, }, ] self._setup_chain(chains_dir, "chain-567-test", tasks) args = argparse.Namespace(task_id="task-567.1") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-567.1.md").write_text("작업 완료.", encoding="utf-8") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" assert "task_id" in output assert output["task_id"] is None def test_next_advances_by_task_id(self, cm, tmp_path, chains_dir, capsys): """create로 생성한 체인에서 task_id로 next 진행이 성공한다.""" tasks = [ {"task_id": "task-300.1", "order": 1, "task_file": "memory/tasks/t300-1.md", "team": "dev1-team"}, {"task_id": "task-300.2", "order": 2, "task_file": "memory/tasks/t300-2.md", "team": "dev2-team"}, ] args_create = argparse.Namespace( chain_id="next-advance-test", tasks=json.dumps(tasks), scope="next 진행 테스트", created_by="anu", max_tasks=10, ) with patch.object(cm, "subprocess") as mock_sub: mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = '{"cron_id": "cron-advance"}' mock_sub.run.return_value = mock_result cm.cmd_create(args_create) # task-300.1을 running으로 변경 args_update = argparse.Namespace(task_id="task-300.1", status="running") cm.cmd_update(args_update) capsys.readouterr() # create/update 출력 비우기 # 보고서 생성 (FAIL 없음) reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-300.1.md").write_text("작업 완료.", encoding="utf-8") # next 호출 args_next = argparse.Namespace(task_id="task-300.1") cm.cmd_next(args_next) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" assert output["task_file"] == "memory/tasks/t300-2.md" assert output["team"] == "dev2-team" # 체인 파일에서 task-300.1이 done으로 마킹되었는지 확인 chain_file = chains_dir / "chain-next-advance-test.json" data = json.loads(chain_file.read_text(encoding="utf-8")) done_task = next(t for t in data["tasks"] if t.get("task_id") == "task-300.1") assert done_task["status"] == "done" assert done_task["completed_at"] is not None # --------------------------------------------------------------------------- # 9. TestF12RetryPhase: F12 completion-promise retry phase 테스트 # --------------------------------------------------------------------------- class TestF12RetryPhase: """F12: completion-promise retry phase 테스트""" def _setup_chain_with_retry( self, chains_dir: Path, tmp_path: Path, task_id: str = "task-f12.1", retry_count: int = 0, report_content: str = "QC FAIL: 타입 에러 발견.", ) -> Path: """F12 테스트용 체인 + 보고서를 생성하는 헬퍼.""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-f12-1.md", "team": "dev2-team", "status": "running", "task_id": task_id, "gate": "auto", "started_at": datetime.now().isoformat(), "completed_at": None, "retry_count": retry_count, }, { "order": 2, "task_file": "memory/tasks/task-f12-2.md", "team": "dev2-team", "status": "pending", "task_id": "task-f12.2", "gate": "auto", "started_at": None, "completed_at": None, }, ] data = { "chain_id": "chain-f12-test", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "F12 retry 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-f12", "tasks": tasks, } path = chains_dir / "chain-f12-test.json" path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / f"{task_id}.md").write_text(report_content, encoding="utf-8") return path def test_f12_first_qc_fail_retries(self, cm, tmp_path, chains_dir, capsys): """retry_count=0인 상태에서 QC FAIL → action=dispatch, retry_attempt=1.""" self._setup_chain_with_retry(chains_dir, tmp_path, retry_count=0) args = argparse.Namespace(task_id="task-f12.1") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" assert output["retry_attempt"] == 1 # 체인 상태는 active 유지 data = json.loads((chains_dir / "chain-f12-test.json").read_text(encoding="utf-8")) assert data["status"] == "active" # target_task status=running, retry_count=1 target_task = next(t for t in data["tasks"] if t.get("task_id") == "task-f12.1") assert target_task["status"] == "running" assert target_task["retry_count"] == 1 def test_f12_second_qc_fail_retries(self, cm, tmp_path, chains_dir, capsys): """retry_count=1인 상태에서 QC FAIL → action=dispatch, retry_attempt=2.""" self._setup_chain_with_retry(chains_dir, tmp_path, retry_count=1) args = argparse.Namespace(task_id="task-f12.1") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" assert output["retry_attempt"] == 2 # 체인 상태는 active 유지 data = json.loads((chains_dir / "chain-f12-test.json").read_text(encoding="utf-8")) assert data["status"] == "active" # target_task retry_count=2 target_task = next(t for t in data["tasks"] if t.get("task_id") == "task-f12.1") assert target_task["retry_count"] == 2 def test_f12_circuit_breaker_on_max_retry(self, cm, tmp_path, chains_dir, capsys): """retry_count=2(=MAX_RETRY)인 상태에서 QC FAIL → circuit breaker 발동.""" self._setup_chain_with_retry(chains_dir, tmp_path, retry_count=2) args = argparse.Namespace(task_id="task-f12.1") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "stalled" assert "circuit_breaker" in output["reason"] assert "escalation_file" in output # 체인 상태가 stalled로 변경 data = json.loads((chains_dir / "chain-f12-test.json").read_text(encoding="utf-8")) assert data["status"] == "stalled" # escalation 파일이 실제 생성되었는지 확인 escalation_file = tmp_path / "memory" / "escalations" / "task-f12.1_escalation.json" assert escalation_file.exists(), f"escalation 파일이 생성되어야 함: {escalation_file}" def test_f12_escalation_file_content(self, cm, tmp_path, chains_dir, capsys): """circuit breaker 발동 후 escalation 파일 내용 검증.""" self._setup_chain_with_retry(chains_dir, tmp_path, retry_count=2) args = argparse.Namespace(task_id="task-f12.1") cm.cmd_next(args) capsys.readouterr() # 출력 비우기 escalation_file = tmp_path / "memory" / "escalations" / "task-f12.1_escalation.json" assert escalation_file.exists() content = json.loads(escalation_file.read_text(encoding="utf-8")) # 필수 필드 검증 assert "task_id" in content assert "chain_id" in content assert "triggered_at" in content assert "reason" in content assert "max_retry" in content assert "total_attempts" in content assert "action" in content assert content["task_id"] == "task-f12.1" assert content["chain_id"] == "chain-f12-test" assert content["max_retry"] == 2 def test_f12_max_retry_constant_is_2(self, cm, tmp_path, chains_dir): """MAX_RETRY == 2 확인.""" assert cm.MAX_RETRY == 2 def test_f12_qc_pass_no_retry_needed(self, cm, tmp_path, chains_dir, capsys): """retry_count=1인 상태에서 QC PASS → action=dispatch (다음 task로 진행).""" self._setup_chain_with_retry( chains_dir, tmp_path, retry_count=1, report_content="작업 완료. 모든 테스트 통과.", # FAIL 없음 ) args = argparse.Namespace(task_id="task-f12.1") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) assert output["action"] == "dispatch" assert output["task_id"] == "task-f12.2" # 다음 task로 진행 # retry_attempt 필드 없음 assert "retry_attempt" not in output def test_f12_retry_resets_task_status(self, cm, tmp_path, chains_dir, capsys): """retry 후 target_task의 status=running, completed_at=None, started_at 갱신 확인.""" self._setup_chain_with_retry(chains_dir, tmp_path, retry_count=0) args = argparse.Namespace(task_id="task-f12.1") before = datetime.now().isoformat() cm.cmd_next(args) capsys.readouterr() data = json.loads((chains_dir / "chain-f12-test.json").read_text(encoding="utf-8")) target_task = next(t for t in data["tasks"] if t.get("task_id") == "task-f12.1") assert target_task["status"] == "running" assert target_task["completed_at"] is None assert target_task["started_at"] is not None # started_at이 before 이후여야 함 (갱신된 것) assert target_task["started_at"] >= before def test_f12_gate_none_bypasses_retry(self, cm, tmp_path, chains_dir, capsys): """gate=none인 task는 QC FAIL이어도 retry 없이 다음 task로 진행.""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-f12-gate-none.md", "team": "dev2-team", "status": "running", "task_id": "task-f12-gate.1", "gate": "none", # gate=none "started_at": datetime.now().isoformat(), "completed_at": None, "retry_count": 0, }, { "order": 2, "task_file": "memory/tasks/task-f12-gate-none-2.md", "team": "dev2-team", "status": "pending", "task_id": "task-f12-gate.2", "gate": "none", "started_at": None, "completed_at": None, }, ] data = { "chain_id": "chain-f12-gate-none", "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "gate=none 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-f12-gate", "tasks": tasks, } chain_file = chains_dir / "chain-f12-gate-none.json" chain_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") # 보고서에 FAIL이 있어도 gate=none이면 retry 없이 다음 task로 진행 reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-f12-gate.1.md").write_text("QC FAIL: 오류 발견.", encoding="utf-8") args = argparse.Namespace(task_id="task-f12-gate.1") cm.cmd_next(args) captured = capsys.readouterr() output = json.loads(captured.out) # gate=none이면 retry 없이 다음 task로 dispatch assert output["action"] == "dispatch" assert output["task_id"] == "task-f12-gate.2" assert "retry_attempt" not in output # --------------------------------------------------------------------------- # 10. TestCircuitBreakerIntegration: circuit_breaker 모듈 통합 테스트 # --------------------------------------------------------------------------- class TestCircuitBreakerIntegration: """circuit_breaker 모듈 통합 테스트 (task-1651.1).""" def test_trigger_uses_circuit_breaker_module(self, tmp_path): """_trigger_circuit_breaker가 circuit_breaker 모듈의 _write_escalation_file을 호출하는지 검증.""" cm = _load_chain_manager(tmp_path) escalations_dir = tmp_path / "memory" / "escalations" cm._trigger_circuit_breaker("task-test-1", "chain-test-1", 3) escalation_file = escalations_dir / "task-test-1_escalation.json" assert escalation_file.exists(), "escalation 파일이 생성되어야 한다" data = json.loads(escalation_file.read_text()) assert data["task_id"] == "task-test-1" assert data["chain_id"] == "chain-test-1" assert data["action"] == "escalation" assert "MAX_RETRY" in data["reason"] def test_trigger_fallback_when_cb_unavailable(self, tmp_path): """circuit_breaker 모듈 미사용 시 fallback으로 escalation 파일이 생성되는지 검증.""" cm = _load_chain_manager(tmp_path) escalations_dir = tmp_path / "memory" / "escalations" # _CB_AVAILABLE을 False로 설정하여 fallback 경로 테스트 original_cb = getattr(cm, "_CB_AVAILABLE", False) setattr(cm, "_CB_AVAILABLE", False) try: cm._trigger_circuit_breaker("task-fallback-1", "chain-fb-1", 3) finally: setattr(cm, "_CB_AVAILABLE", original_cb) escalation_file = escalations_dir / "task-fallback-1_escalation.json" assert escalation_file.exists(), "fallback으로 escalation 파일이 생성되어야 한다" data = json.loads(escalation_file.read_text()) assert data["task_id"] == "task-fallback-1" assert data["chain_id"] == "chain-fb-1" def test_write_escalation_file_extra_fields(self, tmp_path): """_write_escalation_file이 **extra 필드를 payload에 포함하는지 검증.""" import importlib import utils.circuit_breaker as cb_mod importlib.reload(cb_mod) # ESCALATIONS_DIR을 tmp_path로 변경 original_dir = cb_mod.ESCALATIONS_DIR cb_mod.ESCALATIONS_DIR = tmp_path / "escalations" try: cb_mod._write_escalation_file( context="test_extra", reason="test reason", error_count=3, chain_id="chain-extra", custom_field="custom_value", ) finally: cb_mod.ESCALATIONS_DIR = original_dir # 생성된 파일 확인 files = list((tmp_path / "escalations").glob("*_escalation.json")) assert len(files) == 1 data = json.loads(files[0].read_text()) assert data["chain_id"] == "chain-extra" assert data["custom_field"] == "custom_value" assert data["context"] == "test_extra" assert data["error_count"] == 3