cfe2901a55
- itsm/ -> workspace/guardia-itsm/ - manager/ -> workspace/guardia-manager/ - app/ -> workspace/guardia-messenger/ - manual/ -> workspace/guardia-docs/ workspace/zioinfo-web/ unchanged. git mv preserves full commit history. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
239 lines
9.3 KiB
Python
239 lines
9.3 KiB
Python
"""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)
|