""" test_dispatch_workflow.py dispatch.py의 --workflow 옵션과 image_workflow.py의 build_workflow_overview_prompt() 함수 테스트 (아르고스 작성) 테스트 항목: - T1: build_workflow_overview_prompt() 기본 동작 (비어있지 않은 문자열, "Phase", "QC" 포함) - T2: build_workflow_overview_prompt() campaign_type="brand" 파라미터 - T3: dispatch.py argparse에서 --workflow 옵션 인식 - T4: --workflow image-qc-gate 사용 시 task_desc에 워크플로우 프롬프트 prepend 확인 - T5: --workflow 미사용 시 기존 동작 동일 (task_desc 변경 없음) - T6: build_workflow_overview_prompt() QC 기준 정보 포함 (A-01/글자 겹침, 에스컬레이션) """ import json import os import subprocess import sys import types from pathlib import Path from unittest.mock import MagicMock, patch import pytest # --------------------------------------------------------------------------- # 헬퍼: dispatch 모듈을 격리된 WORKSPACE로 재로드 # --------------------------------------------------------------------------- def _load_dispatch_with_workspace(tmp_path: Path) -> types.ModuleType: """dispatch 모듈을 tmp_path를 WORKSPACE로 설정하여 로드한다. test_dispatch.py의 _load_dispatch_with_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 # --------------------------------------------------------------------------- # fixture: tmp_path 기반 dispatch 모듈 # --------------------------------------------------------------------------- @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) return _load_dispatch_with_workspace(tmp_path) # --------------------------------------------------------------------------- # T1: build_workflow_overview_prompt() 기본 동작 # --------------------------------------------------------------------------- class TestBuildWorkflowOverviewPromptBasic: """T1: build_workflow_overview_prompt() 기본 동작 검증""" def test_returns_non_empty_string(self): """task_id='test-001'로 호출 시 비어있지 않은 문자열 반환""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") assert isinstance(result, str) assert len(result) > 0 def test_contains_phase_string(self): """반환값에 'Phase' 문자열 포함""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") assert "Phase" in result def test_contains_qc_string(self): """반환값에 'QC' 문자열 포함""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") assert "QC" in result # --------------------------------------------------------------------------- # T2: build_workflow_overview_prompt() campaign_type 파라미터 # --------------------------------------------------------------------------- class TestBuildWorkflowOverviewPromptCampaignType: """T2: campaign_type 파라미터 동작 검증""" def test_brand_campaign_type_included_in_result(self): """task_id만으로 호출 시 비어있지 않은 문자열 반환 (campaign_type 파라미터 제거됨)""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") # 반환 문자열이 비어있지 않아야 함 assert isinstance(result, str) assert len(result) > 0 def test_default_campaign_type_is_conversion(self): """기본 campaign_type은 'conversion'이어야 함 (명시 없이 호출 가능)""" from prompts.image_workflow import build_workflow_overview_prompt # 기본값으로 호출 가능하고 에러 없이 문자열 반환 result = build_workflow_overview_prompt(task_id="test-001") assert isinstance(result, str) assert len(result) > 0 def test_conversion_campaign_type_included_in_result(self): """task_id만으로 호출 시 Phase 및 QC 내용 포함 (campaign_type 파라미터 제거됨)""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") assert "Phase" in result assert "QC" in result # --------------------------------------------------------------------------- # T3: dispatch.py argparse에서 --workflow 옵션 인식 # --------------------------------------------------------------------------- class TestDispatchWorkflowArgparse: """T3: --workflow 옵션이 argparse에 등록되어 있는지 검증""" def test_help_output_contains_workflow_option(self): """subprocess로 --help 실행 시 '--workflow' 포함 확인""" workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) dispatch_py = workspace / "dispatch.py" result = subprocess.run( [sys.executable, str(dispatch_py), "--help"], capture_output=True, text=True, timeout=30, ) # --help는 exit code 0으로 종료됨 combined_output = result.stdout + result.stderr assert "--workflow" in combined_output, ( f"dispatch.py --help 출력에 '--workflow'가 없음.\n" f"stdout: {result.stdout[:500]}\n" f"stderr: {result.stderr[:500]}" ) def test_workflow_choices_include_image_qc_gate(self): """--help 출력에 'image-qc-gate' 선택지 포함 확인""" workspace = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) dispatch_py = workspace / "dispatch.py" result = subprocess.run( [sys.executable, str(dispatch_py), "--help"], capture_output=True, text=True, timeout=30, ) combined_output = result.stdout + result.stderr assert "image-qc-gate" in combined_output, ( f"dispatch.py --help 출력에 'image-qc-gate'가 없음.\n" f"stdout: {result.stdout[:500]}" ) # --------------------------------------------------------------------------- # T4: --workflow image-qc-gate 사용 시 task_desc에 워크플로우 프롬프트 prepend # --------------------------------------------------------------------------- class TestDispatchWorkflowPrepend: """T4: --workflow image-qc-gate 사용 시 task_desc 앞에 워크플로우 프롬프트가 prepend되는지 검증""" 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_workflow_prompt_prepended_when_workflow_set(self, dispatch_mod, monkeypatch, capsys): """--workflow image-qc-gate 전달 시 task_desc 앞에 워크플로우 개요 프롬프트가 prepend됨""" from prompts.image_workflow import build_workflow_overview_prompt original_task = "이미지 QC 테스트 작업" captured_task_descs = [] original_dispatch = dispatch_mod.dispatch def mock_dispatch(*args, **kwargs): captured_task_descs.append(kwargs.get("task_desc", args[1] if len(args) > 1 else "")) return { "status": "dispatched", "task_id": "task-1.1", "team": "dev1-team", "lead": "헤르메스", "level": "normal", "description": original_task, "message": "ok", "cron_response": {}, } monkeypatch.setattr(dispatch_mod, "dispatch", mock_dispatch) monkeypatch.setattr( sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", original_task, "--workflow", "image-qc-gate"], ) import io from contextlib import redirect_stdout f = io.StringIO() with redirect_stdout(f): dispatch_mod.main() assert len(captured_task_descs) == 1, "dispatch()가 정확히 1회 호출되어야 함" actual_task_desc = captured_task_descs[0] # 워크플로우 프롬프트가 prepend 되어 있어야 함 assert original_task in actual_task_desc, f"원래 task_desc가 유지되어야 함. 실제: {actual_task_desc[:200]}" # 워크플로우 프롬프트(Phase, QC)가 task_desc 앞에 있어야 함 assert ( "Phase" in actual_task_desc ), f"워크플로우 프롬프트의 'Phase'가 task_desc에 없음. 실제: {actual_task_desc[:200]}" assert ( "QC" in actual_task_desc ), f"워크플로우 프롬프트의 'QC'가 task_desc에 없음. 실제: {actual_task_desc[:200]}" # original_task가 워크플로우 프롬프트 뒤에 위치해야 함 (prepend 확인) phase_idx = actual_task_desc.find("Phase") task_idx = actual_task_desc.find(original_task) assert ( phase_idx < task_idx ), f"워크플로우 프롬프트('Phase' at {phase_idx})가 원래 task_desc(at {task_idx}) 앞에 있어야 함" def test_workflow_prompt_uses_correct_task_id(self, dispatch_mod, monkeypatch): """build_workflow_overview_prompt가 실제 task_id를 전달받아 호출됨""" from prompts.image_workflow import build_workflow_overview_prompt original_task = "워크플로우 task_id 확인 테스트" captured_task_descs = [] def mock_dispatch(*args, **kwargs): captured_task_descs.append(kwargs.get("task_desc", args[1] if len(args) > 1 else "")) return { "status": "dispatched", "task_id": "task-1.1", "team": "dev1-team", "lead": "헤르메스", "level": "normal", "description": original_task, "message": "ok", "cron_response": {}, } monkeypatch.setattr(dispatch_mod, "dispatch", mock_dispatch) monkeypatch.setattr( sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", original_task, "--workflow", "image-qc-gate"], ) import io from contextlib import redirect_stdout f = io.StringIO() with redirect_stdout(f): dispatch_mod.main() assert len(captured_task_descs) == 1 actual_task_desc = captured_task_descs[0] # task_desc에 워크플로우 관련 내용이 포함되어야 함 assert len(actual_task_desc) > len( original_task ), "워크플로우 옵션 사용 시 task_desc가 원래보다 길어야 함 (prepend 됨)" # --------------------------------------------------------------------------- # T5: --workflow 미사용 시 기존 동작 동일 # --------------------------------------------------------------------------- class TestDispatchNoWorkflow: """T5: --workflow 미사용 시 task_desc가 변경되지 않음을 검증""" def test_no_workflow_task_desc_unchanged(self, dispatch_mod, monkeypatch): """--workflow 없이 dispatch 호출 시 task_desc가 그대로 전달됨""" original_task = "워크플로우 미사용 기본 작업" captured_task_descs = [] def mock_dispatch(*args, **kwargs): captured_task_descs.append(kwargs.get("task_desc", args[1] if len(args) > 1 else "")) return { "status": "dispatched", "task_id": "task-1.1", "team": "dev1-team", "lead": "헤르메스", "level": "normal", "description": original_task, "message": "ok", "cron_response": {}, } monkeypatch.setattr(dispatch_mod, "dispatch", mock_dispatch) monkeypatch.setattr( sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", original_task], ) import io from contextlib import redirect_stdout f = io.StringIO() with redirect_stdout(f): dispatch_mod.main() assert len(captured_task_descs) == 1 actual_task_desc = captured_task_descs[0] # --workflow 미사용 시 task_desc는 원래 값과 동일해야 함 assert actual_task_desc == original_task, ( f"--workflow 미사용 시 task_desc 변경 없어야 함.\n" f"기대: '{original_task}'\n" f"실제: '{actual_task_desc}'" ) def test_no_workflow_means_no_phase_or_qc_injection(self, dispatch_mod, monkeypatch): """--workflow 없을 때 task_desc에 'Phase'나 'QC' 주입이 없음""" original_task = "일반 코딩 작업 - Phase나 QC 없음" captured_task_descs = [] def mock_dispatch(*args, **kwargs): captured_task_descs.append(kwargs.get("task_desc", args[1] if len(args) > 1 else "")) return { "status": "dispatched", "task_id": "task-1.1", "team": "dev1-team", "lead": "헤르메스", "level": "normal", "description": original_task, "message": "ok", "cron_response": {}, } monkeypatch.setattr(dispatch_mod, "dispatch", mock_dispatch) monkeypatch.setattr( sys, "argv", ["dispatch.py", "--team", "dev1-team", "--task", original_task], ) import io from contextlib import redirect_stdout f = io.StringIO() with redirect_stdout(f): dispatch_mod.main() assert len(captured_task_descs) == 1 actual_task_desc = captured_task_descs[0] # 워크플로우 미사용 시 내용 변경 없음 (원본 그대로) assert actual_task_desc == original_task # --------------------------------------------------------------------------- # T6: build_workflow_overview_prompt()에 QC 기준 정보 포함 # --------------------------------------------------------------------------- class TestBuildWorkflowOverviewPromptQCInfo: """T6: 반환값에 카테고리 A 관련 내용과 에스컬레이션 관련 내용 포함 검증""" def test_qc_details_not_inlined_in_overview(self): """QC 상세(A-01, 글자 겹침 등)는 overview에 포함하지 않음 (Phase별 로딩)""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") assert "A-01" not in result, "QC 상세 A-01이 overview에 인라인되면 안 됨" assert "글자 겹침" not in result, "QC 상세 '글자 겹침'이 overview에 인라인되면 안 됨" def test_qc_loading_reference_present(self): """QC 기준은 각 Phase에서 로딩된다는 안내 포함""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") assert "QC 기준" in result, "'QC 기준' 안내가 없음" assert "각 Phase에서 로딩" in result, "'각 Phase에서 로딩' 안내가 없음" def test_contains_escalation_info(self): """반환값에 에스컬레이션 관련 내용 포함""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") # 에스컬레이션 관련 단어 중 하나 이상 포함 assert "에스컬레이션" in result, f"'에스컬레이션' 텍스트가 없음. 반환값 앞 500자: {result[:500]}" def test_contains_five_phase_structure(self): """반환값에 5Phase (Phase 0~4) 워크플로우 구조 포함""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-001") # 5Phase 구조 전체가 포함되어야 함 (Phase 0부터 Phase 4까지) for phase_num in range(5): assert ( f"Phase {phase_num}" in result or f"phase{phase_num}" in result.lower() ), f"'Phase {phase_num}'가 반환값에 없음. 반환값 앞 500자: {result[:500]}" def test_contains_task_id_in_result(self): """반환값에 전달된 task_id 포함""" from prompts.image_workflow import build_workflow_overview_prompt result = build_workflow_overview_prompt(task_id="test-qc-999") assert "test-qc-999" in result, f"task_id 'test-qc-999'가 반환값에 없음. 반환값 앞 500자: {result[:500]}"