1e98f0d04a
- 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>
104 lines
3.9 KiB
Python
104 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Export/Import 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()
|
|
|
|
test_code = r"""
|
|
import urllib.request, json, io, zipfile
|
|
|
|
BASE = 'http://localhost:8001'
|
|
body = json.dumps({'username':'admin','password':'1111'}).encode()
|
|
req = urllib.request.Request(BASE+'/api/auth/login', data=body, method='POST')
|
|
req.add_header('Content-Type','application/json')
|
|
TOKEN = json.loads(urllib.request.urlopen(req,timeout=10).read()).get('access_token','')
|
|
print('Token:', TOKEN[:20]+'...')
|
|
|
|
def api(method, path, token=TOKEN):
|
|
req = urllib.request.Request(BASE+path, method=method)
|
|
req.add_header('Authorization','Bearer '+TOKEN)
|
|
try:
|
|
r = urllib.request.urlopen(req, timeout=15)
|
|
return json.loads(r.read()), r.status
|
|
except urllib.error.HTTPError as e:
|
|
try: return json.loads(e.read()), e.code
|
|
except: return {}, e.code
|
|
|
|
RESULTS = []
|
|
def test(name, fn):
|
|
try:
|
|
r = fn(); ok = True
|
|
except Exception as ex:
|
|
r = str(ex); ok = False
|
|
RESULTS.append((name, ok, str(r)[:80]))
|
|
print(('OK ' if ok else 'FAIL') + ' ' + name + ': ' + str(r)[:80])
|
|
return r
|
|
|
|
print()
|
|
print('=== Export/Import API 테스트 ===')
|
|
|
|
test('T1 SR Export', lambda: api('GET','/api/export-import/export/sr?limit=5'))
|
|
test('T2 CMDB Export', lambda: api('GET','/api/export-import/export/cmdb'))
|
|
test('T3 기관 Export', lambda: api('GET','/api/export-import/export/institutions'))
|
|
test('T4 감사 로그 Export', lambda: api('GET','/api/export-import/export/audit?limit=5'))
|
|
|
|
def t5_bundle():
|
|
req = urllib.request.Request(BASE+'/api/export-import/export/bundle', method='GET')
|
|
req.add_header('Authorization','Bearer '+TOKEN)
|
|
r = urllib.request.urlopen(req, timeout=20)
|
|
data = r.read()
|
|
with zipfile.ZipFile(io.BytesIO(data)) as z:
|
|
names = z.namelist()
|
|
return 'HTTP '+str(r.status)+' ZIP='+str(len(data)//1024)+'KB files='+str(names)
|
|
test('T5 번들 ZIP Export', t5_bundle)
|
|
|
|
def t6_dry_run():
|
|
sr_data = json.dumps({'data':[{'sr_id':'DRYTEST-001','title':'Dry Run Test SR',
|
|
'status':'RECEIVED','priority':'LOW'}]}).encode()
|
|
boundary = b'testboundary123'
|
|
form = (b'--'+boundary+b'\r\n'
|
|
+b'Content-Disposition: form-data; name="file"; filename="test.json"\r\n'
|
|
+b'Content-Type: application/json\r\n\r\n'
|
|
+sr_data+b'\r\n--'+boundary+b'--\r\n')
|
|
req = urllib.request.Request(BASE+'/api/export-import/import/sr?dry_run=true',
|
|
data=form, method='POST')
|
|
req.add_header('Content-Type','multipart/form-data; boundary=testboundary123')
|
|
req.add_header('Authorization','Bearer '+TOKEN)
|
|
r = urllib.request.urlopen(req, timeout=10)
|
|
d = json.loads(r.read())
|
|
return 'HTTP '+str(r.status)+' status='+str(d.get('status'))+' total='+str(d.get('total'))
|
|
test('T6 SR Import dry_run', t6_dry_run)
|
|
|
|
r7 = urllib.request.urlopen('http://localhost:8090/export-import', timeout=5)
|
|
test('T7 Manager UI /export-import', lambda: 'HTTP '+str(r7.status))
|
|
|
|
print()
|
|
print('='*55)
|
|
passed = sum(1 for _,ok,_ in RESULTS if ok)
|
|
print('결과: '+str(passed)+'/'+str(len(RESULTS))+' PASS')
|
|
for name,ok,detail in RESULTS:
|
|
print((' OK ' if ok else ' FAIL') + ' ' + name)
|
|
print('='*55)
|
|
"""
|
|
|
|
with sftp.open('/tmp/test_ei.py', 'w') as f:
|
|
f.write(test_code)
|
|
sftp.close()
|
|
|
|
chan = client.get_transport().open_session()
|
|
chan.set_combine_stderr(True)
|
|
chan.exec_command('python3 /tmp/test_ei.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.2)
|
|
while chan.recv_ready(): sys.stdout.buffer.write(chan.recv(8192))
|
|
sys.stdout.flush()
|
|
chan.recv_exit_status()
|
|
client.close()
|