zioinfo-web/deploy/test_work_order.py

#!/usr/bin/env python3
"""
Messenger → ITSM 작업지시 송수신 테스트

흐름:
1. Messenger 앱에서 SR(작업지시) 등록
2. ITSM에서 SR 상태 변경 (승인 → 진행 → 완료)
3. Messenger 앱으로 상태 변경 알림 수신 (WebSocket)
4. AI 챗봇으로 자연어 작업지시 → SR 자동 생성
"""
import paramiko, time, sys

HOST='101.79.17.164'; USER='root'; PASS='1q2w3e!Q'
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(HOST, username=USER, password=PASS, timeout=15)
sftp = client.open_sftp()

script = r"""
import http.client, json, sys, time, secrets

HOST = 'localhost'; PORT = 8001
RESULTS = []
SR_ID = None  # 테스트에서 생성한 SR ID 저장

def api(method, path, data=None, token=None):
    conn = http.client.HTTPConnection(HOST, PORT, timeout=10)
    headers = {'Content-Type':'application/json'}
    if token: headers['Authorization'] = 'Bearer ' + token
    body = json.dumps(data).encode() if data else None
    conn.request(method, path, body=body, headers=headers)
    r = conn.getresponse(); raw = r.read()
    try: return json.loads(raw), r.status
    except: return {'raw': raw.decode()[:300]}, r.status

def test(name, fn):
    try:
        r = fn(); ok = True
    except Exception as e: r = str(e)[:100]; ok = False
    RESULTS.append((name, ok, str(r)[:100]))
    print(('PASS' if ok else 'FAIL') + ' ' + name + ': ' + str(r)[:90])
    return r if ok else None

print()
print('='*65)
print('Messenger -> ITSM 작업지시 송수신 테스트')
print('='*65)

# 로그인
d, s = api('POST','/api/auth/login',{'username':'admin','password':'1111'})
TOKEN = d.get('access_token','')
if not TOKEN: print('로그인 실패'); sys.exit(1)
print(f'  로그인 OK (token={TOKEN[:15]}...)')
print()

# ── 단계 1: Messenger 앱에서 작업지시 등록 ─────────────────────────────
print('[단계 1] Messenger 앱 → ITSM 작업지시 등록')

suffix = secrets.token_hex(3).upper()

def step1_sr():
    global SR_ID
    # Messenger 앱의 SR 등록 화면에서 직접 POST
    d, s = api('POST','/api/tasks', {
        'title':       f'[Messenger] 웹서버 재시작 요청 {suffix}',
        'description': 'GUARDiA Messenger 앱에서 직접 등록한 작업지시입니다.\n'
                       '서버: web-01.zioinfo.co.kr\n'
                       '작업: nginx 재시작\n'
                       '요청자: admin (Messenger 앱)',
        'priority':    'HIGH',
        'sr_type':     'RESTART',
        'requested_by': 'admin',
    }, token=TOKEN)
    if s in [200,201]:
        SR_ID = d.get('id') or d.get('sr_id')
        return f'HTTP {s} SR등록 성공: id={d.get("id")} sr_id={d.get("sr_id","?")} title={d.get("title","?")[:30]}'
    return f'HTTP {s}: {d.get("detail","?")[:80]}'

test('S1-1 Messenger→ITSM SR 등록 (POST /api/tasks)', step1_sr)

# SR 목록에서 방금 생성한 SR 확인
def step1_verify():
    d, s = api('GET','/api/tasks?size=1',token=TOKEN)
    if isinstance(d,list) and len(d)>0:
        latest = d[0]
        return f'HTTP {s} 최신SR: {latest.get("sr_id","?")} [{latest.get("status","?")}] {latest.get("title","?")[:30]}'
    return f'HTTP {s} {type(d)}'
test('S1-2 ITSM SR 목록 확인', step1_verify)

print()

# ── 단계 2: ITSM에서 SR 상태 변경 ──────────────────────────────────────
print('[단계 2] ITSM SR 상태 변경 → Messenger 알림 수신')

def step2_approve():
    # SR을 APPROVED 상태로 변경 (승인)
    if not SR_ID: return 'SR_ID 없음'
    # PATCH /api/tasks/{id}/status 또는 PUT
    d, s = api('PATCH', f'/api/tasks/{SR_ID}/status', {'status':'APPROVED'}, token=TOKEN)
    if s in [200,201]:
        return f'HTTP {s} 상태변경: APPROVED -> WebSocket 알림 발생'
    # 다른 경로 시도
    d2, s2 = api('PUT', f'/api/tasks/{SR_ID}', {'status':'IN_PROGRESS'}, token=TOKEN)
    return f'HTTP {s} / {s2}: {d.get("detail","?")} / {d2.get("status","?")}'
test('S2-1 SR 승인 (APPROVED) → 앱 알림 수신', step2_approve)

def step2_inprogress():
    if not SR_ID: return 'SR_ID 없음'
    d, s = api('PATCH', f'/api/tasks/{SR_ID}/status', {'status':'IN_PROGRESS'}, token=TOKEN)
    if s in [200,201]:
        return f'HTTP {s} 상태변경: IN_PROGRESS -> 앱에 진행중 알림'
    return f'HTTP {s}: {d.get("detail","?")[:60]}'
test('S2-2 SR 진행 (IN_PROGRESS) → 앱 알림 수신', step2_inprogress)

def step2_complete():
    if not SR_ID: return 'SR_ID 없음'
    d, s = api('PATCH', f'/api/tasks/{SR_ID}/status', {'status':'COMPLETED'}, token=TOKEN)
    if s in [200,201]:
        return f'HTTP {s} 상태변경: COMPLETED -> 앱에 완료 알림'
    return f'HTTP {s}: {d.get("detail","?")[:60]}'
test('S2-3 SR 완료 (COMPLETED) → 앱 알림 수신', step2_complete)

print()

# ── 단계 3: AI 챗봇으로 작업지시 ──────────────────────────────────────
print('[단계 3] Messenger AI 챗봇 → 자연어 작업지시')

def step3_ai_cmd():
    d, s = api('POST','/api/chatbot/message',
        {'message': '웹서버 nginx를 재시작해 주세요. 긴급합니다.'},
        token=TOKEN)
    reply = d.get('reply', d.get('message', d.get('response', str(d)[:80])))
    return f'HTTP {s} AI응답: {str(reply)[:80]}'
test('S3-1 AI챗봇 자연어 작업지시 (nginx 재시작)', step3_ai_cmd)

def step3_ai_sr():
    d, s = api('POST','/api/chatbot/message',
        {'message': 'SR 등록해줘: 데이터베이스 백업 실행 요청, 우선순위 HIGH'},
        token=TOKEN)
    reply = d.get('reply', d.get('message', str(d)[:80]))
    return f'HTTP {s} AI응답: {str(reply)[:80]}'
test('S3-2 AI챗봇 SR 자동 생성 요청', step3_ai_sr)

def step3_nlcmd():
    # 자연어 명령 (nlcmd)
    d, s = api('POST','/api/ai-cmd',
        {'message': '현재 서버 상태를 확인해줘', 'channel': 'messenger'},
        token=TOKEN)
    return f'HTTP {s}: {str(d)[:80]}'
test('S3-3 AI 자연어 명령 (서버 상태 확인)', step3_nlcmd)

print()

# ── 단계 4: WebSocket 이벤트 직접 확인 ─────────────────────────────────
print('[단계 4] WebSocket 이벤트 상태 확인')

def step4_ws():
    d, s = api('GET','/api/ws/status',token=TOKEN)
    conns = d.get('total_connections', d.get('connection_count', 0))
    channels = d.get('channels', [])
    return f'HTTP {s} 연결={conns}개 채널={channels[:4]}'
test('S4-1 WebSocket 연결 상태 (앱 연결 수)', step4_ws)

def step4_audit():
    d, s = api('GET','/api/audit?size=5',token=TOKEN)
    items = d if isinstance(d,list) else d.get('items',d.get('content',[]))
    logs = [(i.get('action','?'), i.get('username','?')) for i in items[:3]]
    return f'HTTP {s} 최근감사: {logs}'
test('S4-2 감사 로그 (작업지시 이력 추적)', step4_audit)

print()
print('='*65)
passed = sum(1 for _,ok,_ in RESULTS if ok)
print(f'결과: {passed}/{len(RESULTS)} PASS')
print()
for name,ok,d in RESULTS:
    print(f'  {"OK  " if ok else "FAIL"} {name}')
print()
print('='*65)
print()
print('[작업지시 전체 흐름]')
print()
print('  Messenger 앱                    GUARDiA ITSM')
print('       |                                |')
print('  SR 등록 탭에서 입력 ──────────────> POST /api/tasks')
print('       |                                |')
print('  AI챗봇: "nginx 재시작해줘" ─────> /api/chatbot/message')
print('       |                                | (AI가 SR 자동 생성)')
print('       |  <── WebSocket 상태 알림 ──── 관리자 승인')
print('       |  <── WebSocket 완료 알림 ──── 작업 완료 처리')
print()
print('[앱에서 확인하는 방법]')
print('1. 앱 SR탭 -> "새 SR" 버튼 -> 제목/설명/우선순위 입력 -> 등록')
print('   → ITSM 관리자 웹(http://101.79.17.164:8001)에서 SR 확인')
print()
print('2. 앱 AI채팅탭 -> "서버 nginx 재시작 요청해줘" 입력')
print('   → AI가 자동으로 SR 생성 후 응답')
print()
print('3. 앱 알림탭 -> ⚡ 실시간 연결 초록 표시 확인')
print('   → ITSM에서 SR 상태 변경 시 즉시 알림 수신')
"""

with sftp.open('/tmp/test_wo.py','w') as f: f.write(script)
sftp.close()

chan = client.get_transport().open_session()
chan.set_combine_stderr(True)
chan.exec_command('python3 /tmp/test_wo.py 2>&1')
start = time.time()
while not chan.exit_status_ready():
    if chan.recv_ready(): sys.stdout.buffer.write(chan.recv(8192)); sys.stdout.flush()
    if time.time()-start > 50: break
    time.sleep(0.3)
while chan.recv_ready(): sys.stdout.buffer.write(chan.recv(8192))
sys.stdout.flush()
chan.recv_exit_status()
client.close()