""" test_phase4_integration.py fireauto Phase 4 통합 테스트 — F12 completion-promise + 기존 시스템 회귀 검증 테스트 시나리오: 1. chain_manager retry → dispatch → retry → dispatch → circuit breaker 전체 흐름 2. escalation 파일 + audit-trail 기록 동시 검증 3. 기존 chain_manager 기능 회귀 없음 4. 주요 스크립트 존재 확인 (smoke test) """ import argparse import json import os import sys import types from datetime import datetime 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] 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 _create_retry_chain( chains_dir: Path, tmp_path: Path, task_id: str = "task-e2e.1", retry_count: int = 0, report_content: str = "QC FAIL: 오류.", chain_id: str = "chain-e2e-test", ) -> Path: """E2E 테스트용 체인과 보고서를 생성하는 헬퍼.""" tasks = [ { "order": 1, "task_file": "memory/tasks/task-e2e-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-e2e-2.md", "team": "dev2-team", "status": "pending", "task_id": "task-e2e.2", "gate": "auto", "started_at": None, "completed_at": None, }, ] data = { "chain_id": chain_id, "created_by": "anu", "created_at": datetime.now().isoformat(), "status": "active", "scope": "E2E retry 테스트", "max_tasks": 10, "watchdog_cron_id": "cron-e2e", "tasks": tasks, } path = chains_dir / f"{chain_id}.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 # --------------------------------------------------------------------------- # E2E 통합 테스트 # --------------------------------------------------------------------------- class TestE2ERetryFlow: """F12 retry 전체 흐름 E2E 테스트""" def test_e2e_retry_flow_full_cycle(self, cm, tmp_path, chains_dir, capsys): """retry 0→1→2→circuit breaker 전체 3회 시나리오.""" chain_id = "chain-e2e-full" task_id = "task-e2e-full.1" # --- 1회차: retry_count=0, QC FAIL → dispatch (retry_attempt=1) --- _create_retry_chain( chains_dir, tmp_path, task_id=task_id, retry_count=0, chain_id=chain_id, ) cm.cmd_next(argparse.Namespace(task_id=task_id)) output1 = json.loads(capsys.readouterr().out) assert output1["action"] == "dispatch" assert output1["retry_attempt"] == 1 # 체인 상태 active, retry_count=1 확인 chain_file = chains_dir / f"{chain_id}.json" data1 = json.loads(chain_file.read_text(encoding="utf-8")) assert data1["status"] == "active" t1 = next(t for t in data1["tasks"] if t.get("task_id") == task_id) assert t1["retry_count"] == 1 # --- 2회차: retry_count=1로 직접 설정, QC FAIL → dispatch (retry_attempt=2) --- # 보고서 재작성 (같은 FAIL 내용) (tmp_path / "memory" / "reports" / f"{task_id}.md").write_text( "QC FAIL: 타입 에러.", encoding="utf-8" ) cm.cmd_next(argparse.Namespace(task_id=task_id)) output2 = json.loads(capsys.readouterr().out) assert output2["action"] == "dispatch" assert output2["retry_attempt"] == 2 data2 = json.loads(chain_file.read_text(encoding="utf-8")) assert data2["status"] == "active" t2 = next(t for t in data2["tasks"] if t.get("task_id") == task_id) assert t2["retry_count"] == 2 # --- 3회차: retry_count=2(=MAX_RETRY) → circuit breaker --- (tmp_path / "memory" / "reports" / f"{task_id}.md").write_text( "QC FAIL: 여전히 실패.", encoding="utf-8" ) cm.cmd_next(argparse.Namespace(task_id=task_id)) output3 = json.loads(capsys.readouterr().out) assert output3["action"] == "stalled" assert "circuit_breaker" in output3["reason"] # 체인 stalled 확인 data3 = json.loads(chain_file.read_text(encoding="utf-8")) assert data3["status"] == "stalled" # escalation 파일 확인 escalation_file = tmp_path / "memory" / "escalations" / f"{task_id}_escalation.json" assert escalation_file.exists() def test_e2e_retry_then_success(self, cm, tmp_path, chains_dir, capsys): """retry 1회 후 QC PASS → 다음 task 정상 dispatch.""" chain_id = "chain-e2e-success" task_id = "task-e2e-success.1" # 1회차: QC FAIL → retry _create_retry_chain( chains_dir, tmp_path, task_id=task_id, retry_count=0, chain_id=chain_id, ) cm.cmd_next(argparse.Namespace(task_id=task_id)) out_retry = json.loads(capsys.readouterr().out) assert out_retry["action"] == "dispatch" assert out_retry["retry_attempt"] == 1 # 체인 파일에서 retry_count=1 확인 chain_file = chains_dir / f"{chain_id}.json" data = json.loads(chain_file.read_text(encoding="utf-8")) t = next(t for t in data["tasks"] if t.get("task_id") == task_id) assert t["retry_count"] == 1 # 2회차: QC PASS (FAIL 없는 보고서) → 다음 task dispatch (tmp_path / "memory" / "reports" / f"{task_id}.md").write_text( "작업 완료. 모든 테스트 통과.", encoding="utf-8" ) cm.cmd_next(argparse.Namespace(task_id=task_id)) out_success = json.loads(capsys.readouterr().out) assert out_success["action"] == "dispatch" assert out_success["task_id"] == "task-e2e.2" # 다음 task assert "retry_attempt" not in out_success # 체인 상태 active 유지 final_data = json.loads(chain_file.read_text(encoding="utf-8")) assert final_data["status"] == "active" # --------------------------------------------------------------------------- # Smoke Test: 주요 스크립트 존재 확인 # --------------------------------------------------------------------------- class TestScriptsExistSmoke: """주요 스크립트 파일 존재 smoke test""" def test_scripts_exist_smoke(self): """주요 스크립트 파일 존재 확인.""" scripts = { "dispatch.py": Path("/home/jay/workspace/dispatch.py"), "chain_manager.py": Path("/home/jay/workspace/chain_manager.py"), "notify-completion.py": Path("/home/jay/workspace/scripts/notify-completion.py"), "finish-task.sh": Path("/home/jay/workspace/scripts/finish-task.sh"), "task-timer.py": Path("/home/jay/workspace/memory/task-timer.py"), "qc_verify.py": Path("/home/jay/workspace/teams/shared/qc_verify.py"), "whisper-compile.py": Path("/home/jay/workspace/scripts/whisper-compile.py"), "memory-search.py": Path("/home/jay/workspace/scripts/memory-search.py"), } missing = [] for name, path in scripts.items(): if not path.exists(): missing.append(f"{name} ({path})") assert not missing, f"누락된 스크립트: {', '.join(missing)}" # --------------------------------------------------------------------------- # 기존 chain_manager 기능 회귀 테스트 # --------------------------------------------------------------------------- class TestChainManagerNoRegression: """기존 chain_manager 기능 회귀 없음 확인""" def test_chain_manager_no_regression(self, cm, tmp_path, chains_dir, capsys): """기존 cmd_create, cmd_update, cmd_check 정상 동작 확인.""" # --- cmd_create 회귀 --- tasks = [ { "order": 1, "task_file": "memory/tasks/regress-1.md", "team": "dev1-team", "gate": "auto", "task_id": "task-regress.1", }, { "order": 2, "task_file": "memory/tasks/regress-2.md", "team": "dev2-team", "gate": "auto", "task_id": "task-regress.2", }, ] args_create = argparse.Namespace( chain_id="regression-chain", 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-regress"}' mock_sub.run.return_value = mock_result cm.cmd_create(args_create) chain_file = chains_dir / "chain-regression-chain.json" assert chain_file.exists(), "체인 파일이 생성되어야 함" data = json.loads(chain_file.read_text(encoding="utf-8")) assert data["chain_id"] == "regression-chain" assert data["status"] == "active" assert len(data["tasks"]) == 2 capsys.readouterr() # create 출력 비우기 # --- cmd_update 회귀: task-regress.1 → running --- args_update = argparse.Namespace(task_id="task-regress.1", status="running") cm.cmd_update(args_update) updated_data = json.loads(chain_file.read_text(encoding="utf-8")) t1 = next(t for t in updated_data["tasks"] if t.get("task_id") == "task-regress.1") assert t1["status"] == "running" assert t1["started_at"] is not None capsys.readouterr() # --- cmd_check 회귀: task-regress.1 체인 소속 확인 --- args_check = argparse.Namespace(task_id="task-regress.1") cm.cmd_check(args_check) captured = capsys.readouterr() check_output = json.loads(captured.out) assert check_output["in_chain"] is True assert check_output["chain_id"] == "regression-chain" assert check_output["is_last"] is False assert check_output["next_task_id"] == "task-regress.2" # --- cmd_next 회귀: QC PASS → 다음 task dispatch --- reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True, exist_ok=True) (reports_dir / "task-regress.1.md").write_text("작업 완료. 성공.", encoding="utf-8") args_next = argparse.Namespace(task_id="task-regress.1") cm.cmd_next(args_next) captured = capsys.readouterr() next_output = json.loads(captured.out) assert next_output["action"] == "dispatch" assert next_output["task_id"] == "task-regress.2" assert next_output["team"] == "dev2-team" # 체인 상태 active 유지 final_data = json.loads(chain_file.read_text(encoding="utf-8")) assert final_data["status"] == "active"