YPE html>

#主程序
from flask import Flask
from flask import request
import subprocess
import json


htmltemp="""

      

        
压力测试利器！
        
利用测试平台的web功能来达到快速控制后端Python Locust服务器，填写需要的测试数据通过后台运行Locust实例在线完成测试的调试工作，解决测试组工作电脑性能不足的问题.
        
Learn more »
      
    

    

      
      

        

          
脚本配置
          
  
          
host
    
          
测试URL
    
          
提交方法
    
          GET
POST
附加参数
    
          提交配置 
          
 
        
       

         
      


正在运行
          
运行地址
          
操作
        

实例1
          
http://10.101.0.xx:xxxx
          
关闭Locust后台进程
        

  
       
        
      

""" def do_mk_pyfile(): testurl=request.form[‘test-url‘] meth = request.form[‘meth‘] if request.form[‘addp‘]: addp="\"" + str(request.form[‘addp‘]) + "\"" ss = "sh /opt/web-server/locustpy.sh {meth} {url} {addp}".format(url=testurl, meth=meth, addp=addp) else: ss = "sh /opt/web-server/locustpy.sh {meth} {url} ".format(url=testurl, meth=meth) subprocess.Popen(ss,shell=True) sdata={} app = Flask(__name__) @app.route(‘/‘) def index(): return htmltemp @app.route(‘/locust‘, methods=[‘POST‘]) def locust(): print(request.form[‘addp‘]) do_mk_pyfile() ss = "nohup locust -f /opt/web-server/usecase.py --host={host} --port=xxxx 2>&1 &".format(host=request.form[‘host‘]) subprocess.Popen(ss,shell=True) sdata[‘status‘]=1 return resphtml @app.route(‘/status‘) def status(): return sdata @app.route(‘/closelocust‘, methods=[‘POST‘]) def closeLocust(): sdata[‘status‘]=0 ss = "ps aux |grep locust|grep -v grep|awk ‘{print $2}‘|xargs kill -9" subprocess.Popen(ss,shell=True) return "OK" if __name__ == ‘__main__‘: app.debug = True # 设置调试模式，生产模式的时候要关掉debug app.run(host="10.101.0.xx", debug=True)

shell脚本

#!/bin/bash


echo "" > /opt/web-server/usecase.py
method=$1
taskurl=$2
echo $3

if [ $method == "GET" ];then
cat  /opt/web-server/usecase.py
from locust import HttpLocust, TaskSet, task

class MyTaskSet(TaskSet):
    @task()
    def test(self):
        self.client.get("${taskurl}")

class MyLocust(HttpLocust):
    task_set = MyTaskSet
    min_wait = 5000
    max_wait = 15000
EOF
else
if [ ! -n "${3}" ]; then 
cat  /opt/web-server/usecase.py
from locust import HttpLocust, TaskSet, task

class MyTaskSet(TaskSet):
    @task()
    def test(self):
        self.client.post("${taskurl}")

class MyLocust(HttpLocust):
    task_set = MyTaskSet
    min_wait = 3000
    max_wait = 5000
EOF
else
cat  /opt/web-server/usecase.py
from locust import HttpLocust, TaskSet, task

class MyTaskSet(TaskSet):
    @task()
    def test(self):
        self.client.post("${taskurl}",${3})

class MyLocust(HttpLocust):
    task_set = MyTaskSet
    min_wait = 3000
    max_wait = 5000
EOF
fi
fi

locust脚本：

from locust import HttpLocust, TaskSet, task

class MyTaskSet(TaskSet):
    @task()
    def test(self):
        self.client.get("")

class MyLocust(HttpLocust):
    task_set = MyTaskSet
    min_wait = 5000
    max_wait = 15000

python locust 压力测试web 利用flask编写

标签：mes   navig   temp   eth   sign   call   hup   contain   page   

原文地址：https://www.cnblogs.com/kuku0223/p/12859226.html