11c670f2a0
- 37개 파일 IP → zioinfo.co.kr 치환 (소스/매뉴얼/설정/하네스) - Manager DrConsole/NetworkConsole/CsapConsole 빌드 + /var/www/manager/ 배포 - 테스트: Manager HTTP 200, ITSM 신규 API 7개 전체 200 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""서버에서 직접 실행되는 알림 테스트 스크립트"""
|
|
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
|
|
|
|
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)[:80]))
|
|
print(('PASS' if ok else 'FAIL') + ' ' + name + ': ' + str(r)[:80])
|
|
return r if ok else None
|
|
|
|
print()
|
|
print('='*60)
|
|
print('GUARDiA ITSM -> Messenger 알림 송수신 테스트')
|
|
print('='*60)
|
|
|
|
# T1: 로그인
|
|
d, s = api('POST', '/api/auth/login', {'username':'admin','password':'1111'})
|
|
TOKEN = d.get('access_token','')
|
|
test('T1 관리자 로그인', lambda: 'OK token=' + TOKEN[:15] + '...' if TOKEN else 'FAIL: ' + str(d))
|
|
|
|
# T2: WebSocket 연결 수 확인
|
|
test('T2 WebSocket 상태', lambda: str(api('GET','/api/ws/status',token=TOKEN)))
|
|
|
|
# T3: SR 등록 → 이벤트 발생
|
|
import hashlib; sr_suffix = hashlib.md5(str(time.time()).encode()).hexdigest()[:6].upper()
|
|
def t3():
|
|
d, s = api('POST','/api/tasks', {
|
|
'title': f'[알림테스트] Messenger 연동 SR-{sr_suffix}',
|
|
'description': 'GUARDiA Messenger WebSocket 알림 연동 테스트. 앱 알림탭에서 확인하세요.',
|
|
'priority':'MEDIUM', 'sr_type':'OTHER'}, token=TOKEN)
|
|
return f'HTTP {s} sr_id={d.get("sr_id","?")} title={d.get("title","?")[:30]}'
|
|
test('T3 SR 등록 (WebSocket 이벤트 트리거)', t3)
|
|
|
|
# T4: 알림 로그 조회
|
|
def t4():
|
|
d, s = api('GET','/api/notifications/log?size=5',token=TOKEN)
|
|
cnt = len(d) if isinstance(d,list) else '?'
|
|
return f'HTTP {s} count={cnt}'
|
|
test('T4 알림 로그 조회', t4)
|
|
|
|
# T5: 메신저 알림 테스트
|
|
def t5():
|
|
d, s = api('POST','/api/notifications/test-messenger',token=TOKEN)
|
|
return f'HTTP {s}: {d.get("message",d.get("detail",str(d)[:50]))}'
|
|
test('T5 메신저 알림 테스트 API', t5)
|
|
|
|
# T6: 이메일 알림 테스트 (ythong86@gmail.com)
|
|
def t6():
|
|
d, s = api('POST','/api/notifications/test-email',
|
|
{'email':'ythong86@gmail.com','subject':'GUARDiA Messenger 알림 테스트',
|
|
'message':'GUARDiA Messenger 앱이 정상적으로 ITSM과 연동되었습니다.'},
|
|
token=TOKEN)
|
|
return f'HTTP {s}: {d.get("message",str(d)[:50])}'
|
|
test('T6 이메일 알림 발송 (ythong86@gmail.com)', t6)
|
|
|
|
# T7: 대시보드 API (앱 메인화면 연동)
|
|
def t7():
|
|
d, s = api('GET','/api/dashboard',token=TOKEN)
|
|
keys = list(d.keys())[:4]
|
|
return f'HTTP {s} keys={keys}'
|
|
test('T7 대시보드 API (앱 연동)', t7)
|
|
|
|
# T8: SR 목록 조회 (앱 SR탭 연동)
|
|
def t8():
|
|
d, s = api('GET','/api/tasks?size=3',token=TOKEN)
|
|
cnt = d.get('total_elements', d.get('total', '?'))
|
|
return f'HTTP {s} total={cnt}'
|
|
test('T8 SR 목록 API (앱 SR탭 연동)', t8)
|
|
|
|
print()
|
|
print('='*60)
|
|
passed = sum(1 for _,ok,_ in RESULTS if ok)
|
|
print(f'결과: {passed}/{len(RESULTS)} PASS')
|
|
for name,ok,d in RESULTS:
|
|
print(('OK ' if ok else 'FAIL') + ' ' + name)
|
|
print('='*60)
|
|
print()
|
|
print('[Messenger 앱 알림 확인 방법]')
|
|
print('1. 앱 실행 -> 알림 탭')
|
|
print('2. 상단: "GUARDiA 실시간 연결" 초록 표시 확인')
|
|
print('3. SR 등록 즉시 -> 알림 목록에 [실시간] 배지로 표시')
|
|
print('4. 배포 실행 시 -> 배포 완료/실패 알림 자동 수신')
|
|
"""
|
|
|
|
with sftp.open('/tmp/notif_test.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_test.py 2>&1')
|
|
start = time.time()
|
|
while not chan.exit_status_ready():
|
|
if chan.recv_ready(): sys.stdout.buffer.write(chan.recv(4096)); sys.stdout.flush()
|
|
if time.time()-start > 30: break
|
|
time.sleep(0.3)
|
|
while chan.recv_ready(): sys.stdout.buffer.write(chan.recv(4096))
|
|
sys.stdout.flush()
|
|
chan.recv_exit_status()
|
|
client.close()
|