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>
152 lines
5.0 KiB
Python
152 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
GUARDiA ITSM → GUARDiA Messenger 알림 송수신 테스트
|
|
|
|
테스트 항목:
|
|
T1. 관리자 로그인 (JWT 발급)
|
|
T2. WebSocket 연결 테스트 (ws://localhost:8001/ws/events)
|
|
T3. SR 등록 → 이벤트 발생 확인
|
|
T4. 알림 로그 조회 (REST API)
|
|
T5. 메신저 알림 테스트 엔드포인트
|
|
T6. Messenger 앱 알림 API 조회
|
|
"""
|
|
import paramiko, time, sys, json
|
|
|
|
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()
|
|
|
|
def run(label, cmd, timeout=20):
|
|
print(f'\n[{label}]')
|
|
chan = client.get_transport().open_session()
|
|
chan.set_combine_stderr(True)
|
|
chan.exec_command(cmd)
|
|
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 > timeout: print('[TIMEOUT]'); break
|
|
time.sleep(0.3)
|
|
while chan.recv_ready(): sys.stdout.buffer.write(chan.recv(4096))
|
|
sys.stdout.flush()
|
|
chan.recv_exit_status()
|
|
|
|
test_script = """
|
|
import urllib.request, urllib.error, json, base64, time, sys
|
|
|
|
BASE = 'http://localhost:8001'
|
|
RESULTS = []
|
|
|
|
def api(method, path, data=None, token=None, form=False):
|
|
url = BASE + path
|
|
if form:
|
|
import urllib.parse
|
|
body = urllib.parse.urlencode(data).encode() if data else None
|
|
ct = 'application/x-www-form-urlencoded'
|
|
else:
|
|
body = json.dumps(data).encode() if data else None
|
|
ct = 'application/json'
|
|
req = urllib.request.Request(url, data=body, method=method)
|
|
req.add_header('Content-Type', ct)
|
|
if token: req.add_header('Authorization', 'Bearer '+token)
|
|
try:
|
|
r = urllib.request.urlopen(req, timeout=10)
|
|
return json.loads(r.read()), r.status
|
|
except urllib.error.HTTPError as e:
|
|
try: return json.loads(e.read()), e.code
|
|
except: return {'error': str(e)}, e.code
|
|
|
|
def test(name, fn):
|
|
try:
|
|
r = fn(); ok = True
|
|
except Exception as e:
|
|
r = str(e); ok = False
|
|
RESULTS.append((name, ok, str(r)[:100]))
|
|
icon = 'PASS' if ok else 'FAIL'
|
|
print(f' [{icon}] {name}: {str(r)[:80]}')
|
|
return r if ok else None
|
|
|
|
print()
|
|
print('='*60)
|
|
print('GUARDiA 알림 송수신 테스트')
|
|
print('='*60)
|
|
|
|
# T1: 로그인
|
|
d, s = api('POST', '/api/auth/login', {'username':'admin','password':'1111'})
|
|
TOKEN = d.get('access_token','')
|
|
test('T1 관리자 로그인', lambda: f'HTTP {s}, Token={TOKEN[:15]}...' if TOKEN else f'FAIL: {d}')
|
|
|
|
if not TOKEN:
|
|
print('토큰 없음 - 테스트 중단')
|
|
sys.exit(1)
|
|
|
|
# T2: WebSocket 상태 확인
|
|
def t2():
|
|
d, s = api('GET', '/api/ws/status', token=TOKEN)
|
|
return f'HTTP {s}, connections={d.get(\"connection_count\",d)}'
|
|
test('T2 WebSocket 상태', t2)
|
|
|
|
# T3: SR 등록 (이벤트 발생)
|
|
def t3():
|
|
import secrets
|
|
sr_id = f'TEST-{secrets.token_hex(3).upper()}'
|
|
d, s = api('POST', '/api/tasks', {
|
|
'title': f'[알림 테스트] SR {sr_id}',
|
|
'description': 'GUARDiA Messenger 알림 연동 테스트용 SR',
|
|
'priority': 'MEDIUM',
|
|
'sr_type': 'OTHER',
|
|
}, token=TOKEN)
|
|
return f'HTTP {s}, sr_id={d.get(\"sr_id\",d.get(\"detail\",\"?\"))}'
|
|
test('T3 SR 등록 (이벤트 트리거)', t3)
|
|
|
|
# T4: 알림 로그 조회
|
|
def t4():
|
|
d, s = api('GET', '/api/notifications/log?size=5', token=TOKEN)
|
|
cnt = len(d) if isinstance(d,list) else d.get('total',d.get('count','?'))
|
|
items = d[:2] if isinstance(d,list) else d.get('items',d.get('content',[]))[:2]
|
|
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)}'
|
|
test('T5 메신저 알림 테스트', t5)
|
|
|
|
# T6: SSE 이벤트 버스 상태
|
|
def t6():
|
|
d, s = api('GET', '/api/dashboard', token=TOKEN)
|
|
return f'HTTP {s}: dashboard ok, keys={list(d.keys())[:3]}'
|
|
test('T6 대시보드 API (앱 연동 확인)', t6)
|
|
|
|
# T7: 알림 설정 조회
|
|
def t7():
|
|
d, s = api('GET', '/api/notifications/config', token=TOKEN)
|
|
return f'HTTP {s}: smtp={d.get(\"smtp_enabled\",\"?\")}, messenger={d.get(\"messenger_enabled\",\"?\")}'
|
|
test('T7 알림 설정 조회', t7)
|
|
|
|
print()
|
|
print('='*60)
|
|
passed = sum(1 for _,ok,_ in RESULTS if ok)
|
|
print(f'결과: {passed}/{len(RESULTS)} PASS')
|
|
for name,ok,detail in RESULTS:
|
|
print(f' {\"OK \" if ok else \"FAIL\"} {name}')
|
|
print('='*60)
|
|
|
|
print()
|
|
print('[앱 연동 확인]')
|
|
print('GUARDiA Messenger 앱에서:')
|
|
print('1. 로그인 → 알림 탭 탭')
|
|
print('2. 상단에 "GUARDiA 실시간 연결" 초록 표시 확인')
|
|
print('3. SR 등록 또는 배포 실행 시 알림 자동 표시')
|
|
print('4. ⚡ 실시간 태그로 WebSocket 수신 구분')
|
|
"""
|
|
|
|
with sftp.open('/tmp/test_notif.py', 'w') as f:
|
|
f.write(test_script)
|
|
sftp.close()
|
|
|
|
run('알림 테스트 실행', 'python3 /tmp/test_notif.py 2>&1', 30)
|
|
client.close()
|