302 lines
12 KiB
Python
302 lines
12 KiB
Python
"""GUARDiA ITSM 단위 테스트 — tenant_ai + workflow_engine"""
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
|
|
|
|
import json
|
|
import pytest
|
|
from datetime import datetime
|
|
|
|
|
|
# ── TenantAIModel / TenantKBDoc ORM 모델 ────────────────────────────────────
|
|
|
|
class TestTenantAIModelORM:
|
|
def test_import_orm_models(self):
|
|
from models import TenantAIModel, TenantKBDoc
|
|
assert TenantAIModel.__tablename__ == "tb_tenant_ai_model"
|
|
assert TenantKBDoc.__tablename__ == "tb_tenant_kb_doc"
|
|
|
|
def test_tenant_ai_model_columns(self):
|
|
from models import TenantAIModel
|
|
cols = {c.name for c in TenantAIModel.__table__.columns}
|
|
required = {"id", "tenant_id", "model_name", "base_model",
|
|
"dataset_size", "status", "accuracy", "created_at"}
|
|
assert required.issubset(cols), f"누락 컬럼: {required - cols}"
|
|
|
|
def test_tenant_kb_doc_columns(self):
|
|
from models import TenantKBDoc
|
|
cols = {c.name for c in TenantKBDoc.__table__.columns}
|
|
required = {"id", "tenant_id", "title", "content", "created_at"}
|
|
assert required.issubset(cols), f"누락 컬럼: {required - cols}"
|
|
|
|
def test_tenant_id_indexed(self):
|
|
from models import TenantAIModel
|
|
indexed_cols = {
|
|
c.name for c in TenantAIModel.__table__.columns if c.index
|
|
}
|
|
assert "tenant_id" in indexed_cols
|
|
|
|
def test_kb_tenant_id_indexed(self):
|
|
from models import TenantKBDoc
|
|
indexed_cols = {
|
|
c.name for c in TenantKBDoc.__table__.columns if c.index
|
|
}
|
|
assert "tenant_id" in indexed_cols
|
|
|
|
|
|
# ── WorkflowDefinition / WorkflowRun ORM 모델 ───────────────────────────────
|
|
|
|
class TestWorkflowORM:
|
|
def test_import_workflow_models(self):
|
|
from models import WorkflowDefinition, WorkflowRun
|
|
assert WorkflowDefinition.__tablename__ == "tb_workflow_definition"
|
|
assert WorkflowRun.__tablename__ == "tb_workflow_run"
|
|
|
|
def test_workflow_definition_columns(self):
|
|
from models import WorkflowDefinition
|
|
cols = {c.name for c in WorkflowDefinition.__table__.columns}
|
|
required = {"id", "name", "trigger", "steps", "active", "created_at"}
|
|
assert required.issubset(cols), f"누락 컬럼: {required - cols}"
|
|
|
|
def test_workflow_run_columns(self):
|
|
from models import WorkflowRun
|
|
cols = {c.name for c in WorkflowRun.__table__.columns}
|
|
required = {"id", "definition_id", "trigger_data", "step_results",
|
|
"status", "started_at", "finished_at"}
|
|
assert required.issubset(cols), f"누락 컬럼: {required - cols}"
|
|
|
|
def test_workflow_run_fk_to_definition(self):
|
|
from models import WorkflowRun
|
|
fk_cols = {fk.column.table.name for fk in WorkflowRun.__table__.foreign_keys}
|
|
assert "tb_workflow_definition" in fk_cols
|
|
|
|
def test_workflow_definition_relationship(self):
|
|
from models import WorkflowDefinition
|
|
# relationship 'runs' 존재 확인
|
|
assert hasattr(WorkflowDefinition, "runs")
|
|
|
|
def test_workflow_run_relationship(self):
|
|
from models import WorkflowRun
|
|
# relationship 'definition' 존재 확인
|
|
assert hasattr(WorkflowRun, "definition")
|
|
|
|
def test_workflow_active_default_false(self):
|
|
from models import WorkflowDefinition
|
|
# active 컬럼 기본값 False
|
|
active_col = WorkflowDefinition.__table__.columns["active"]
|
|
assert active_col.default.arg is False
|
|
|
|
def test_workflow_run_status_default(self):
|
|
from models import WorkflowRun
|
|
status_col = WorkflowRun.__table__.columns["status"]
|
|
assert status_col.default.arg == "running"
|
|
|
|
|
|
# ── tenant_ai 라우터 Pydantic 스키마 ─────────────────────────────────────────
|
|
|
|
class TestTenantAIPydantic:
|
|
def test_train_request_valid(self):
|
|
from routers.tenant_ai import TrainRequest
|
|
req = TrainRequest(model_name="my-llama", base_model="llama3")
|
|
assert req.model_name == "my-llama"
|
|
assert req.base_model == "llama3"
|
|
|
|
def test_train_request_defaults(self):
|
|
from routers.tenant_ai import TrainRequest
|
|
req = TrainRequest(model_name="model-x", base_model="llama3")
|
|
assert req.description is None
|
|
|
|
def test_query_request_valid(self):
|
|
from routers.tenant_ai import QueryRequest
|
|
req = QueryRequest(question="서버 재시작 절차는?")
|
|
assert req.use_kb is True
|
|
assert req.top_k == 3
|
|
|
|
def test_query_request_top_k_limit(self):
|
|
from routers.tenant_ai import QueryRequest
|
|
import pydantic
|
|
with pytest.raises((ValueError, pydantic.ValidationError)):
|
|
QueryRequest(question="질문", top_k=11) # max 10
|
|
|
|
def test_kb_doc_create_valid(self):
|
|
from routers.tenant_ai import KBDocCreate
|
|
doc = KBDocCreate(title="서버 운영 가이드", content="서버 운영 절차...")
|
|
assert doc.title == "서버 운영 가이드"
|
|
|
|
def test_query_request_min_length(self):
|
|
from routers.tenant_ai import QueryRequest
|
|
import pydantic
|
|
with pytest.raises((ValueError, pydantic.ValidationError)):
|
|
QueryRequest(question="") # min_length=1
|
|
|
|
|
|
# ── workflow_engine 라우터 Pydantic 스키마 ───────────────────────────────────
|
|
|
|
class TestWorkflowEnginePydantic:
|
|
def test_workflow_create_valid(self):
|
|
from routers.workflow_engine import WorkflowCreate, WorkflowStep
|
|
req = WorkflowCreate(
|
|
name="테스트 워크플로우",
|
|
trigger={"event": "SR_CREATED"},
|
|
steps=[WorkflowStep(seq=1, type="notify", params={"channel": "messenger"})],
|
|
)
|
|
assert req.name == "테스트 워크플로우"
|
|
assert req.active is False
|
|
assert len(req.steps) == 1
|
|
|
|
def test_workflow_update_partial(self):
|
|
from routers.workflow_engine import WorkflowUpdate
|
|
upd = WorkflowUpdate(active=True)
|
|
assert upd.active is True
|
|
assert upd.name is None
|
|
assert upd.steps is None
|
|
|
|
def test_trigger_request_valid(self):
|
|
from routers.workflow_engine import TriggerRequest
|
|
req = TriggerRequest(definition_id=1, payload={"server_id": "svr-01"})
|
|
assert req.definition_id == 1
|
|
assert req.payload["server_id"] == "svr-01"
|
|
|
|
def test_workflow_create_requires_steps(self):
|
|
from routers.workflow_engine import WorkflowCreate, WorkflowStep
|
|
import pydantic
|
|
with pytest.raises((ValueError, pydantic.ValidationError)):
|
|
WorkflowCreate(name="빈 워크플로우", steps=[]) # min_length=1
|
|
|
|
def test_workflow_step_defaults(self):
|
|
from routers.workflow_engine import WorkflowStep
|
|
step = WorkflowStep(seq=1, type="notify")
|
|
assert step.params == {}
|
|
|
|
|
|
# ── 내장 템플릿 시드 데이터 검증 ────────────────────────────────────────────
|
|
|
|
class TestBuiltinTemplates:
|
|
def test_template_count(self):
|
|
from routers.workflow_engine import BUILTIN_TEMPLATES
|
|
assert len(BUILTIN_TEMPLATES) == 5
|
|
|
|
def test_all_templates_have_required_fields(self):
|
|
from routers.workflow_engine import BUILTIN_TEMPLATES
|
|
for tpl in BUILTIN_TEMPLATES:
|
|
assert "name" in tpl
|
|
assert "description" in tpl
|
|
assert "trigger" in tpl
|
|
assert "steps" in tpl
|
|
|
|
def test_template_names(self):
|
|
from routers.workflow_engine import BUILTIN_TEMPLATES
|
|
names = {tpl["name"] for tpl in BUILTIN_TEMPLATES}
|
|
expected = {
|
|
"SR 자동처리",
|
|
"SLA 에스컬레이션",
|
|
"SSL 인증서 갱신",
|
|
"서버 이상 감지 → SR 생성",
|
|
"정기 보고서 생성",
|
|
}
|
|
assert expected == names
|
|
|
|
def test_all_templates_steps_are_list(self):
|
|
from routers.workflow_engine import BUILTIN_TEMPLATES
|
|
for tpl in BUILTIN_TEMPLATES:
|
|
assert isinstance(tpl["steps"], list)
|
|
assert len(tpl["steps"]) >= 1
|
|
|
|
def test_steps_json_serializable(self):
|
|
from routers.workflow_engine import BUILTIN_TEMPLATES
|
|
for tpl in BUILTIN_TEMPLATES:
|
|
serialized = json.dumps(tpl["steps"])
|
|
parsed = json.loads(serialized)
|
|
assert isinstance(parsed, list)
|
|
|
|
def test_cron_templates_have_cron_expr(self):
|
|
from routers.workflow_engine import BUILTIN_TEMPLATES
|
|
cron_templates = [t for t in BUILTIN_TEMPLATES if t["trigger"].get("event") == "CRON"]
|
|
for tpl in cron_templates:
|
|
assert "cron_expr" in tpl["trigger"], f"{tpl['name']} CRON 트리거에 cron_expr 누락"
|
|
|
|
|
|
# ── _get_tenant_id 헬퍼 ──────────────────────────────────────────────────────
|
|
|
|
class TestTenantIdHelper:
|
|
def test_returns_inst_code_when_present(self):
|
|
from routers.tenant_ai import _get_tenant_id
|
|
|
|
class FakeUser:
|
|
inst_code = "INST001"
|
|
username = "admin"
|
|
|
|
assert _get_tenant_id(FakeUser()) == "INST001"
|
|
|
|
def test_falls_back_to_username(self):
|
|
from routers.tenant_ai import _get_tenant_id
|
|
|
|
class FakeUser:
|
|
inst_code = None
|
|
username = "admin"
|
|
|
|
assert _get_tenant_id(FakeUser()) == "admin"
|
|
|
|
def test_empty_inst_code_falls_back(self):
|
|
from routers.tenant_ai import _get_tenant_id
|
|
|
|
class FakeUser:
|
|
inst_code = ""
|
|
username = "engineer1"
|
|
|
|
# 빈 문자열은 falsy → username 사용
|
|
result = _get_tenant_id(FakeUser())
|
|
assert result == "engineer1"
|
|
|
|
|
|
# ── _execute_step 단위 테스트 ────────────────────────────────────────────────
|
|
|
|
class TestExecuteStep:
|
|
@pytest.mark.asyncio
|
|
async def test_notify_step(self):
|
|
from routers.workflow_engine import _execute_step
|
|
step = {"type": "notify", "params": {"channel": "messenger", "message": "테스트 알림"}}
|
|
result = await _execute_step(step, {}, None)
|
|
assert result["type"] == "notify"
|
|
assert result["result"] == "ok"
|
|
assert result["channel"] == "messenger"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_assign_step(self):
|
|
from routers.workflow_engine import _execute_step
|
|
step = {"type": "auto_assign", "params": {"role": "ENGINEER"}}
|
|
result = await _execute_step(step, {}, None)
|
|
assert result["result"] == "ok"
|
|
assert "ENGINEER" in result["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_escalate_step(self):
|
|
from routers.workflow_engine import _execute_step
|
|
step = {"type": "escalate", "params": {"target_role": "PM"}}
|
|
result = await _execute_step(step, {}, None)
|
|
assert result["result"] == "ok"
|
|
assert result["target"] == "PM"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_step_skipped(self):
|
|
from routers.workflow_engine import _execute_step
|
|
step = {"type": "unknown_action", "params": {}}
|
|
result = await _execute_step(step, {}, None)
|
|
assert result["result"] == "skipped"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_sr_step_formats_title(self):
|
|
from routers.workflow_engine import _execute_step
|
|
step = {"type": "create_sr", "params": {"title": "이상 감지: {server_id}", "priority": "CRITICAL"}}
|
|
result = await _execute_step(step, {"server_id": "svr-99"}, None)
|
|
assert result["result"] == "ok"
|
|
assert "svr-99" in result["title"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_report_step(self):
|
|
from routers.workflow_engine import _execute_step
|
|
step = {"type": "generate_report", "params": {"type": "monthly", "format": "pdf"}}
|
|
result = await _execute_step(step, {}, None)
|
|
assert result["result"] == "ok"
|
|
assert result["report_type"] == "monthly"
|