测试环境数据库带 ssh 通道把相关代码放在 locust 脚本里就会报错没法运行,单独作为一个脚本运行倒是好的
ssh 通道数据库连接部分
with SSHTunnelForwarder(
('13.112.45.12', 22), # B机器的配置
ssh_pkey=flie,
ssh_username="dj",
remote_bind_address=(flie2, 3306)) as server: # A机器的配置
conn = MySQLdb.connect(host='127.0.0.1', # 此处必须是是127.0.0.1
port=server.local_bind_port,
user='user_app',
passwd='51e4dbe0',
charset='utf8')
cursor = conn.cursor()
cursor.execute("SELECT VERSION()")
data = cursor.fetchone()
print "Database version : %s " % data
sql = "select db_member_01.tb_user.uid from db_member_01.tb_user where uid like '8%' GROUP BY auto_id Asc"
cursor.execute(sql)
results = cursor.fetchall()
报错截图