接口API用例自动转locust测试用例

做接口测试是必要的，写完接口测试用例，再写locust压测脚本，其实差异不大：

写个简单的py，把接口测试脚本转为locust压测脚本，本例只是简单的示范：

原接口校验脚本：

 1 # -*- coding = utf-8 -*-
 2 # ------------------------------
 3 # @time: 2020/8/24 15:59
 4 # @Author: drew_gg
 5 # @File: exhibition_exhibit_info.py
 6 # @Software: api_locust
 7 # ------------------------------
 8 
 9 import requests
10 
11 
12 # need to locust #
13 class KBH:
14 
15     host = "https://XXXXer.cn/api"
16     url = "/XXXition/XXibit/info"
17 
18     @classmethod
19     def get_exhibition_exhibit_info(cls):
20         """
21         :return:
22         """
23 
24         data = {"exhibitId": 34}
25         # 请求参数组装 ## r_url:固定参数
26         r_url = KBH.host + KBH.url
27         # 发起请求
28         r = requests.get(r_url, params=data)
29         if r.json()['data']['info']['id'] and r.status_code == 200:
30             print("success")
31         else:
32             print('error')
33 
34 
35 if __name__ == "__main__":
36     KBH().get_exhibition_exhibit_info()

1.代码里需要有一下标识：

（1）@File:

（2）import requests

（3）need to locust

（4）class

（5）url

（6）@classmethod

（7）cls

（8）host

（9）r = requests.

大致这些标识

转换后的代码：

# -*- coding = utf-8 -*-
# ------------------------------
# @time: 2020/8/24 15:59
# @Author: drew_gg
# @File: locust_XXXxhibit_info.py
# @Software: api_locust
# ------------------------------

import requests
from locust import task, constant 
from locust.contrib.fasthttp import FastHttpUser


# need to locust #
class KBH(FastHttpUser):

    host = "https://XXXXer.cn/api"
    url = "/exXXion/XXXibit/info"
    wait_time = constant(1)

    @task
    def get_exhibition_exhibit_info(self):
        """
        :return:
        """

        data = {"exhibitId": 34}
        # 请求参数组装 ## r_url:固定参数
        r_url = KBH.url
        # 发起请求
        with self.client.get(r_url, params=data, timeout=1, catch_response=True) as r:
            if r.content == b"":
                r.failure("No data")
            if r.status_code != 200:
                r.failure("request error")
            print(r.json())

如果有多个压测方法的话，按照这个类似循环修改吧，这里只是做个简单的字符匹配与替换

 1 # -*- coding = utf-8 -*-
 2 # ------------------------------
 3 # @time: 2020/8/21 11:54
 4 # @Author: drew_gg
 5 # @File: case_to_locust.py
 6 # @Software: api_locust
 7 # ------------------------------
 8 
 9 import os
10 
11 pl = os.getcwd().split('api_locust')
12 path_to_do = pl[0] + "api_locust\locust_view\kbh_api\api\"
13 path_to_end = pl[0] + "api_locust\locust_view\kbh_api\locust_api\"
14 
15 
16 def search(path, name):
17     """
18     遍历文档目录
19     :param path:
20     :param name:
21     :return:
22     """
23     file_l = []
24     for root, dirs, files in os.walk(path):
25         root = str(root)
26         if files:
27             for i in files:
28                 if name in i:
29                     if '__init__' not in i:
30                         file_l.append(root + i)
31     return file_l
32 
33 
34 fl = search(path_to_do, '.py')
35 
36 for fi in fl:
37     with open(fi, 'r', encoding="utf-8") as f:
38         py_file = path_to_end + 'locust_' + fi.split('\')[-1]
39         f_new = open(py_file, 'w', encoding='utf-8')
40         f = f.readlines()
41         class_host = '&&&&&&&&@@@'
42         for i in f:
43             if "need to locust" in i:
44                 for line in f:
45                     if "@File:" in line:
46                         b = "# @File: " + 'locust_' + fi.split('\')[-1] + '
'
47                         line = line.replace(line, b)
48                     if "import" in line:
49                         b = line + "from locust import task, constant 
from locust.contrib.fasthttp import FastHttpUser
"
50                         line = line.replace(line, b)
51                     if "class " in line:
52                         b = line.split(":")[0] + "(FastHttpUser):
"
53                         class_name = line.split('class ')[1].split(":")[0]
54                         class_host = class_name + ".host + "
55                         line = line.replace(line, b)
56                     if 'url = "' in line:
57                         b = line + "    wait_time = constant(1)
"
58                         line = line.replace(line, b)
59                     if "@classmethod" in line:
60                         line = line.replace(line, "    @task
")
61                     if "cls" in line:
62                         b = line.split("cls")[0] + 'self' + line.split("cls")[1]
63                         line = line.replace(line, b)
64                     if class_host in line:
65                         b = line.split("KBH.host + ")[0] + line.split("KBH.host + ")[1]
66                         line = line.replace(line, b)
67                     if "r = requests." in line:
68                         r_d = line.split('(')[1].split(")")[0]
69                         r_m = line.split('.')[1].split('(')[0]
70                         if r_m == "get":
71                             b = "        with self.client.get(%s, timeout=1, catch_response=True) as r:
" % r_d
72                         if r_m == "post":
73                             b = "        with self.client.post(%s, timeout=1, catch_response=True) as r:
" % r_d
74                         line = line.replace(line, b)
75                         f_new.write(line)
76                         b = """            if r.content == b"":
77                 r.failure("No data")
78             if r.status_code != 200:
79                 r.failure("request error")
80             print(r.json())
81 """
82                         f_new.write(b)
83                         break
84                     f_new.write(line)
85         f_new.close()

应该有其他更好的方式，欢迎交流