guardia-itsm/test_b5_orchestrator.py

"""B-5 멀티 에이전트 협업 오케스트레이션 테스트"""
import sys, ast, os

os.environ.setdefault("GUARDIA_SECRET_KEY", "test-b5-secret-key-32bytes-padded!")
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_b5.db")

ok = True

print("=== 1. 구문 검사 ===")
files = ["core/orchestrator.py", "routers/orchestrator.py", "main.py"]
for f in files:
    try:
        with open(f, encoding="utf-8") as fh:
            src = fh.read()
        ast.parse(src)
        print(f"  OK  {f}")
    except SyntaxError as e:
        print(f"  ERR {f}: {e}")
        ok = False

print("\n=== 2. models.py WorkflowInstance 확인 ===")
with open("models.py", encoding="utf-8") as f:
    models_src = f.read()

checks = [
    ("WorkflowInstance", "WorkflowInstance ORM 클래스"),
    ("WorkflowStep", "WorkflowStep ORM 클래스"),
    ("WorkflowInstanceOut", "WorkflowInstanceOut Pydantic 스키마"),
    ("WorkflowCreateRequest", "WorkflowCreateRequest Pydantic 스키마"),
    ("tb_workflow_instance", "tb_workflow_instance 테이블명"),
    ("tb_workflow_step", "tb_workflow_step 테이블명"),
    ("workflow_type", "workflow_type 컬럼"),
    ("progress_pct", "progress_pct 컬럼"),
    ("total_steps", "total_steps 컬럼"),
    ("current_step", "current_step 컬럼"),
]
for sym, desc in checks:
    status = "OK" if sym in models_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 3. core/orchestrator.py 함수 및 템플릿 확인 ===")
with open("core/orchestrator.py", encoding="utf-8") as f:
    orch_src = f.read()

orch_checks = [
    ("WORKFLOW_TEMPLATES", "워크플로우 템플릿 딕셔너리"),
    ("SR_TO_DEPLOY", "SR→배포 워크플로우 템플릿"),
    ("INCIDENT_RESP", "인시던트 대응 워크플로우 템플릿"),
    ("CODE_REVIEW", "코드 리뷰 워크플로우 템플릿"),
    ("AGENT_ACTIONS", "에이전트 액션 레지스트리"),
    ("async def _execute_action(", "에이전트 액션 실행 함수"),
    ("async def execute_workflow(", "워크플로우 실행 엔진"),
    ("async def create_workflow_instance(", "워크플로우 인스턴스 생성 함수"),
    ("simulated", "시뮬레이션 모드 (API 미연결 폴백)"),
    ("WorkflowStatus.RUNNING", "RUNNING 상태 전환"),
    ("WorkflowStatus.FAILED", "FAILED 상태 전환"),
    ("WorkflowStatus.COMPLETED", "COMPLETED 상태 전환"),
]
for sym, desc in orch_checks:
    status = "OK" if sym in orch_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 4. routers/orchestrator.py 엔드포인트 확인 ===")
with open("routers/orchestrator.py", encoding="utf-8") as f:
    router_src = f.read()

endpoint_checks = [
    ('@router.post("/workflows"', "POST /api/orchestrator/workflows"),
    ('@router.get("/workflows"', "GET /api/orchestrator/workflows"),
    ('@router.get("/workflows/{instance_id}"', "GET /api/orchestrator/workflows/{id}"),
    ('@router.post("/workflows/{instance_id}/retry"', "POST retry"),
    ('@router.delete("/workflows/{instance_id}"', "DELETE cancel"),
    ('@router.get("/templates"', "GET /api/orchestrator/templates"),
    ('@router.get("/stats"', "GET /api/orchestrator/stats"),
    ("background_tasks", "BackgroundTasks 비동기 실행"),
    ("execute_workflow", "워크플로우 실행 함수 호출"),
    ("WORKFLOW_TEMPLATES", "템플릿 딕셔너리 참조"),
]
for sym, desc in endpoint_checks:
    status = "OK" if sym in router_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 5. main.py 등록 확인 ===")
with open("main.py", encoding="utf-8") as f:
    main_src = f.read()

main_checks = [
    ("orchestrator", "orchestrator 라우터 임포트"),
    ("orchestrator.router", "orchestrator 라우터 등록"),
]
for sym, desc in main_checks:
    status = "OK" if sym in main_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 6. WORKFLOW_TEMPLATES 구조 검증 ===")
try:
    import importlib.util
    spec = importlib.util.spec_from_file_location("orch_mod", "core/orchestrator.py")
    orch_mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(orch_mod)

    templates = orch_mod.WORKFLOW_TEMPLATES
    assert isinstance(templates, dict), "WORKFLOW_TEMPLATES가 dict가 아님"
    assert "SR_TO_DEPLOY" in templates, "SR_TO_DEPLOY 없음"
    assert "INCIDENT_RESP" in templates, "INCIDENT_RESP 없음"
    assert "CODE_REVIEW" in templates, "CODE_REVIEW 없음"
    print(f"  OK  템플릿 수: {len(templates)}")

    for wf_type, steps in templates.items():
        assert isinstance(steps, list) and len(steps) > 0, f"{wf_type} 단계 없음"
        for step in steps:
            assert "order" in step, f"{wf_type} step에 order 없음"
            assert "agent_name" in step, f"{wf_type} step에 agent_name 없음"
            assert "action" in step, f"{wf_type} step에 action 없음"
        print(f"  OK  {wf_type}: {len(steps)}단계, 에이전트={list({s['agent_name'] for s in steps})}")

except AssertionError as e:
    print(f"  ERR  템플릿 구조 오류: {e}")
    ok = False
except Exception as e:
    print(f"  ERR  템플릿 로드 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 7. 에이전트 액션 레지스트리 검증 ===")
try:
    agent_actions = orch_mod.AGENT_ACTIONS
    required_agents = ["sr-manager", "code-reviewer", "deploy-engineer", "kb-agent"]
    for agent in required_agents:
        status = "OK" if agent in agent_actions else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  에이전트: {agent}")

    # 각 에이전트의 액션 출력
    for agent, actions in agent_actions.items():
        print(f"       {agent}: {list(actions.keys())}")

except Exception as e:
    print(f"  ERR  에이전트 레지스트리 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 8. _execute_action 시뮬레이션 테스트 ===")
import asyncio

async def test_execute_action():
    try:
        # 알려진 에이전트/액션 — API 미연결이므로 simulated 모드
        result = await orch_mod._execute_action(
            agent_name="sr-manager",
            action="create_incident_sr",
            context={"sr_id": "SR-TEST-001"},
        )
        assert isinstance(result, dict), "결과가 dict가 아님"
        assert "success" in result, "success 필드 없음"
        assert "data" in result, "data 필드 없음"
        print(f"  OK  _execute_action 반환: success={result['success']}, data={result['data']}")
    except Exception as e:
        print(f"  ERR  _execute_action 오류: {type(e).__name__}: {e}")

asyncio.run(test_execute_action())

print("\n=== 9. WorkflowCreateRequest 검증 ===")
try:
    import importlib.util as ilu
    from typing import Optional, List, Dict
    spec2 = ilu.spec_from_file_location("models_mod", "models.py")
    models_mod = ilu.module_from_spec(spec2)
    # 타이핑 모듈을 models_mod 네임스페이스에 주입
    models_mod.__dict__["Optional"] = Optional
    models_mod.__dict__["List"] = List
    models_mod.__dict__["Dict"] = Dict
    spec2.loader.exec_module(models_mod)

    # 불완전한 모델 rebuild
    for cls_name in ["WorkflowStepOut", "WorkflowInstanceOut", "WorkflowCreateRequest"]:
        cls = getattr(models_mod, cls_name, None)
        if cls and hasattr(cls, "model_rebuild"):
            try:
                cls.model_rebuild()
            except Exception:
                pass

    # WorkflowCreateRequest 필드 확인 (소스 기반)
    with open("models.py", encoding="utf-8") as f:
        ms = f.read()
    req_fields = ["workflow_type", "title", "sr_id", "project_id", "context"]
    # WorkflowCreateRequest 클래스 섹션 찾기
    start = ms.find("class WorkflowCreateRequest(BaseModel):")
    end = ms.find("\n\nclass ", start + 1)
    section = ms[start:end] if end > 0 else ms[start:]
    for field in req_fields:
        status = "OK" if field in section else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  WorkflowCreateRequest.{field}")

    # WorkflowInstanceOut 필드 확인 (소스 기반)
    start2 = ms.find("class WorkflowInstanceOut(BaseModel):")
    end2 = ms.find("\n\nclass ", start2 + 1)
    section2 = ms[start2:end2] if end2 > 0 else ms[start2:]
    required_fields = ["id", "workflow_type", "status", "title", "progress_pct", "total_steps"]
    for field in required_fields:
        status = "OK" if field in section2 else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  WorkflowInstanceOut.{field}")

except Exception as e:
    print(f"  ERR  모델 검증 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 10. CUSTOM 워크플로우 지원 확인 ===")
try:
    # CUSTOM 타입은 WORKFLOW_TEMPLATES에 없어도 허용
    custom_step = {
        "order": 1,
        "agent_name": "sr-manager",
        "action": "custom_action",
        "description": "커스텀 단계",
    }
    assert "CUSTOM" not in orch_mod.WORKFLOW_TEMPLATES, "CUSTOM이 템플릿에 있으면 안 됨"
    print("  OK  CUSTOM 워크플로우는 템플릿 없이 허용 (routers에서 처리)")
except Exception as e:
    print(f"  ERR  CUSTOM 확인 오류: {type(e).__name__}: {e}")

print("\n=== B-5 멀티 에이전트 협업 오케스트레이션 테스트 완료 ===")
if ok:
    print("모든 검사 통과")
else:
    sys.exit(1)