测试环境数据库带 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()

报错截图


↙↙↙阅读原文可查看相关链接,并与作者交流