测试系统中有一项记录ssh登录日志，需要对此进行并发压力测试。
于是用多线程进行python并发记录
因为需要安装的一些依赖和模块比较麻烦，脚本完成后再用pyinstaller打成exe包分发给其他测试人员一起使用。
1.脚本编写
# -*- coding: utf-8 -*-
import paramiko
import threading
import time
lt = []
def ssh(a,xh,sp):
count = 0
for i in range(0,xh):
try:
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('ip地址',22,'用户名', '密码')
ssh.close()
print u"线程[%s]第[%s]次登录"%(a,i)
if sp != 0:
time.sleep(sp)
count += 1
except:
print u"线程异常，已处理"
lt.append(count)
if __name__ == "__main__":
figlet = '''''
_____ _____ _
| ___| | _ \ | |
| |__ | |_| | | |
| __| | _ { | |
| | | |_| | | |
|_| |_____/ |_|
Code by FBI.
'''
print figlet
print u"认证攻击次数=线程数*每个线程认证攻击次数"
print u"请输入线程数:"
xc = raw_input()
print u"请输入每个线程攻击次数:"
xh = raw_input()
print u"请输入每个线程延迟时间(秒)，0为不休眠:"
sp = raw_input()
try:
print u"预计总共发送认证攻击%s次"%(int(xc)*int(xh))
threads = []
for j in range(int(xc)):
threads.append(threading.Thread(target=ssh,args=(j,int(xh),int(sp),)))
for t in threads:
t.start()
print t.name
t.join()
print lt
count = 0
for count in lt:
count += count
print u"程序执行完毕总共发送认证攻击【%s】次" % count
except ValueError,e:
print u"因为输入不规范导致程序出现错误，请输入数字"
2.pyinstaller制作exe程序
下载pyinstaller后
在根目录中cmd中执行python setup.py install安装pyinstaller
安装完成后执行命令打成exe文件
python pyinstaller.py -F 文件路径
3.执行效果
如图：
内容来源于网络如有侵权请私信删除
- 还没有人评论，欢迎说说您的想法！
客服
