zioinfo-web/deploy/notif_test_v2.py

#!/usr/bin/env python3
"""GUARDiA 알림 테스트 v2 - 정확한 API 경로 사용"""
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 = []

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()[:200]}, r.status

def test(name, fn):
    try:
        r = fn(); ok = True
    except Exception as e: r = str(e)[:80]; 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('GUARDiA ITSM -> Messenger 알림 송수신 테스트 v2')
print('='*65)

# T1: 로그인
d, s = api('POST', '/api/auth/login', {'username':'admin','password':'1111'})
TOKEN = d.get('access_token','')
test('T1 관리자 로그인', lambda: f'HTTP {s} OK' if TOKEN else f'FAIL: {d}')

# T2: WebSocket 연결 수 (앱이 연결되면 1 이상)
def t2():
    d, s = api('GET','/api/ws/status',token=TOKEN)
    conns = d.get('total_connections', d.get('connection_count', 0))
    msg = f'연결 수={conns}'
    if conns == 0: msg += ' (앱이 서버에 연결되면 숫자 증가)'
    return f'HTTP {s} {msg}'
test('T2 WebSocket 연결 상태', t2)

# T3: 올바른 SR 등록 API 탐색
def t3():
    # tasks API 확인
    d, s = api('GET','/api/tasks?size=1',token=TOKEN)
    if isinstance(d, list):
        return f'HTTP {s} list형태 SR {len(d)}건'
    elif isinstance(d, dict):
        cnt = d.get('total_elements', d.get('total', d.get('count','?')))
        return f'HTTP {s} total={cnt}'
    return f'HTTP {s} {type(d).__name__}'
test('T3 SR 목록 조회 (앱 SR탭)', t3)

# T4: SR 신규 등록 (올바른 필드)
sr_id_suffix = secrets.token_hex(3).upper()
def t4():
    # GUARDiA ITSM의 nlcmd (자연어 명령) 또는 tasks로 SR 등록
    # 기관 코드 없이 등록 가능한지 확인
    d, s = api('POST','/api/tasks', {
        'title': f'[알림테스트] Messenger 연동 {sr_id_suffix}',
        'description': 'GUARDiA Messenger WebSocket 알림 연동 테스트',
        'priority':'MEDIUM',
        'sr_type':'INQUIRY',
        'inst_id': None,
    }, token=TOKEN)
    if s in [200,201]:
        return f'HTTP {s} SR등록: {d.get("sr_id","?")} -> WebSocket 브로드캐스트 발생'
    return f'HTTP {s}: {d.get("detail","?")[:60]}'
test('T4 SR 등록 (WebSocket 이벤트 트리거)', t4)

# T5: 실시간 이벤트 직접 브로드캐스트 (내부 API)
def t5():
    # SSE/WebSocket 직접 브로드캐스트 테스트
    d, s = api('POST','/api/notifications/test-messenger', token=TOKEN)
    return f'HTTP {s}: {str(d)[:80]}'
test('T5 메신저 알림 테스트 (내부)', t5)

# T6: 알림 로그 확인 (이벤트 기록)
def t6():
    d, s = api('GET','/api/notifications/log?size=10', token=TOKEN)
    if isinstance(d, list):
        return f'HTTP {s} {len(d)}건 로그'
    return f'HTTP {s} {d.get("total","?")}건'
test('T6 알림 로그 확인', t6)

# T7: 감사 로그에서 이벤트 확인
def t7():
    d, s = api('GET','/api/audit?size=3', token=TOKEN)
    if isinstance(d, list) and len(d) > 0:
        latest = d[0]
        return f'HTTP {s} 최신: {latest.get("action","?")} by {latest.get("username","?")}'
    elif isinstance(d, dict):
        items = d.get('items', d.get('content', []))
        if items:
            return f'HTTP {s} 최신: {items[0].get("action","?")} by {items[0].get("username","?")}'
    return f'HTTP {s} {type(d).__name__}'
test('T7 감사 로그 (이벤트 추적)', t7)

# T8: AI 챗봇 API (앱 챗탭 연동)
def t8():
    d, s = api('POST','/api/chatbot/message',
        {'message':'안녕하세요, Messenger 연동 테스트입니다.'},
        token=TOKEN)
    reply = d.get('reply', d.get('message', d.get('response', str(d)[:50])))
    return f'HTTP {s} 응답: {str(reply)[:60]}'
test('T8 AI 챗봇 API (앱 챗탭 연동)', t8)

print()
print('='*65)
passed = sum(1 for _,ok,_ in RESULTS if ok)
print(f'결과: {passed}/{len(RESULTS)} PASS')
print()
for name,ok,detail in RESULTS:
    print(f'  {"OK  " if ok else "FAIL"} {name}')
print()
print('='*65)
print()
print('[앱 알림 동작 흐름]')
print()
print('  GUARDiA ITSM                    GUARDiA Messenger 앱')
print('       |                                  |')
print('  SR 등록 (POST /api/tasks)         WebSocket 연결됨')
print('       |                                  |')
print('  events.broadcast() 호출         이벤트 수신 (lastEvent)')
print('       |                                  |')
print('  ws.broadcast() 호출             알림 탭에 즉시 표시')
print('       |                                  |')
print('  tb_notification_log 저장        ⚡ 실시간 배지 표시')
print()
print('[Messenger 앱에서 확인할 내용]')
print('1. 알림 탭 상단: "GUARDiA 실시간 연결" + 초록 깜빡임')
print('2. SR 등록 즉시: ⚡ 실시간 배지와 함께 알림 항목 추가')
print('3. 배포 실행 시: 🚀 배포 완료 알림 자동 표시')
print('4. 인시던트 발생: 🚨 긴급 알림 즉시 수신')
"""

with sftp.open('/tmp/notif_v2.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/notif_v2.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 > 40: 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()