【Python】使用Beautiful Soup定制Jmeter测试脚本

柔光的暖阳◎ 2023-08-17 15:17 137阅读 0赞

## 背景介绍 ##

我们在做性能调优时，时常需要根据实际压测的情况，调整线程组的参数，比如循环次数，线程数，所有线程启动的时间等。  
如果是在一台Linux机器上，就免不了在本机打开图形页面修改，然后最后传递到压测机上面的过程，所有为了解决这个业务痛点  
，使用Python写了一个能直接修改Jmeter基础压测参数的脚本，能修改jmx脚本的线程组数、循环次数、线程组全部启动需要花的时间。

## 实现思路 ##

刚开始准备写这个脚本的时候，想了两个思路：

#### 把脚本数据读出，使用正则表达式（re库）匹配关键数据进行修改 ####

优点：可以快速的改写数据  
缺点：无法进行区块的修改

#### 把脚本数据读出，使用BeautifulSoup的xml解析功能解析后修改 ####

**注：我们的Jmx脚本其实就是一个标准格式的xml**

**优点：** 能快速的查找元素并进行修改  
**缺点：** 需要熟悉BeautifulSoup的用法

## 通过Beautiful Soup ##

#### Beautiful Soup ####

Beautiful Soup 是一个可以从HTML或XML文件中提取数据的Python库.我们使用BeautifulSoup解析xml或者html的时候，能够得到一个 BeautifulSoup 的对象，我们可以通过操作这个对象来完成原始数据的结构化数据。具体的使用可以参照这份[文档][Link 1]。

#### 具体实现 ####

主要使用了bs4的`soup.find` 和`self.soup.find_all`功能。结化或数据的修改如`loops.string = num`。

值得注意的是，find\_all支持正则匹配，甚至如果没有合适过滤器，那么还可以定义一个方法,方法只接受一个元素参数。

修改后的脚本将以`"T{}L{}R{}-{}_{}.jmx".format(thread_num, loop_num, ramp_time, self.src_script, self.get_time())` 的形式保存，具体封装如下：

import time
    import os
    from bs4 import BeautifulSoup
    
    class OpJmx:
        def __init__(self, file_name):
            self.src_script = self._split_filename(file_name)
            with open(file_name, "r") as f:
                data = f.read()
            self.soup = BeautifulSoup(data, "xml")
    
        @staticmethod
        def _split_filename(filename):
            """
            新生成的文件兼容传入相对路径及文件名称
            :param filename:
            :return:
            """
            relative = filename.split("/")
            return relative[len(relative)-1].split(".jmx")[0]
    
        def _theard_num(self):
            """
            :return: 线程数据对象
            """
            return self.soup.find("stringProp", {"name": {"ThreadGroup.num_threads"}})
    
        def _ramp_time(self):
            """
            :return: 启动所有线程时间配置对象
            """
            return self.soup.find("stringProp", {"name": {"ThreadGroup.ramp_time"}})
    
        def _bean_shell(self):
            """
            :return:  bean_shell对象
            """
            return self.soup.find("stringProp", {"name": {"BeanShellSampler.query"}})
    
        def _paths(self):
            """
            :return: 请求路径信息对象
            """
            return self.soup.find_all("stringProp", {"name": {"HTTPSampler.path"}})
    
        def _methods(self):
            """
            :return: 请求方法对象
            """
            return self.soup.find_all("stringProp", {"name": {"HTTPSampler.method"}})
    
        def _argument(self):
            """
            :return: post请求参数对象
            """
            # Argument.value 不唯一 通过HTTPArgument.always_encode找到
            return self.soup.find_all("boolProp", {"name": {"HTTPArgument.always_encode"}})[0].find_next()
    
        def _loops(self):
            """
            循环次数，兼容forever 与具体次数
            :return: 循环次数对象
            """
            _loops = self.soup.find("stringProp", {"name": {"LoopController.loops"}})
            if _loops:
                pass
            else:
                _loops = self.soup.find("intProp", {"name": {"LoopController.loops"}})
    
            return _loops
    
        @staticmethod
        def get_time():
            return time.strftime("%Y-%m-%d@%X", time.localtime())
    
        def get_bean_shell(self):
            _str = self._bean_shell().string
            logger.info("bean_shell: " + _str)
            return _str
    
        def set_bean_shell(self, new_bean_shell):
            old_bean_shell = self._bean_shell()
            old_bean_shell.string = new_bean_shell
    
        def get_ramp_time(self):
            _str = self._ramp_time().string
            logger.info("ramp_time: " + _str)
            return _str
    
        @check_num
        def set_ramp_time(self, num):
            loops = self._ramp_time()
            loops.string = num
    
        def get_loops(self):
            _str = self._loops().string
            logger.info("loops: " + _str)
            return _str
    
        @check_num
        def set_loops(self, num):
            """
            :param num: -1 为一直循环，其他为具体循环次数
            :return:
            """
            loops = self._loops()
            loops.string = num
    
        def get_argument(self):
            _str = self._argument().string
            logger.info("argument: " + _str)
            return _str
    
        def set_argument(self, **kwargs):
            """
            设置请求参数（JSON，传入字典）
            :param kwargs:
            :return:
            """
            param = self._argument()
            param.string = str(kwargs)
    
        def get_thread_num(self):
            _str = self._theard_num().string
            logger.info("thread_num: " + _str)
            return _str
    
        @check_num
        def set_thread_num(self, num):
            """
            设置线程数信息
            :param num:
            :return:
            """
            thread_num = self._theard_num()
            thread_num.string = num
            # print(self.soup.find_all("stringProp", {"name": {"ThreadGroup.num_threads"}})[0].string)
    
        def mod_header(self, key, value, index=0):
            """
            修改指定header的信息，默认修改第一个值
            :param key:
            :param value:
            :param index:
            :return:
            """
            headers = self.soup.find_all("elementProp", {"elementType": {"Header"}})
            headers[index].find("stringProp", {"name": {"Header.name"}}).string = key
            headers[index].find("stringProp", {"name": {"Header.value"}}).string = value
            # for header in headers:
            #     header.find("stringProp", {"name": {"Header.name"}}).string = key
            #     header.find("stringProp", {"name": {"Header.value"}}).string = value
    
        def save_jmx(self):
            logger.info("参数设置完毕，开始保存数据")
            cur_path = os.path.dirname(os.path.realpath(__file__))
            thread_num = self.get_thread_num()
            loop_num = self.get_loops()
            ramp_time = self.get_ramp_time()
    
            script_name = "T{}L{}R{}-{}_{}.jmx".format(thread_num, loop_num, ramp_time, self.src_script, self.get_time())
            script_path = os.path.join(cur_path, '..', 'script')
    
            if not os.path.exists(script_path):
                os.mkdir(script_path)
    
            script_location = os.path.join(script_path, script_name)
            logger.info("测试脚本已保存于 {}".format(script_location))
            with open(script_location, "w") as f:
                f.write(str(self.soup))
    
            return script_name
    if __name__ == '__main__':
        jmx = OpJmx("templates/template.jmx")
        argvs = sys.argv
        len_argvs = len(argvs) - 1
        if len_argvs == 0:
            pass
        elif len_argvs == 1:
            jmx.set_thread_num(argvs[1])
        elif len_argvs == 2:
            jmx.set_thread_num(argvs[1])
            jmx.set_loops(argvs[2])
        elif len_argvs == 3:
            jmx.set_thread_num(argvs[1])
            jmx.set_loops(argvs[2])
            jmx.set_ramp_time(argvs[3])
        jmx.save_jmx()

未完待续...

转载于:https://www.cnblogs.com/Detector/p/10409879.html

[Link 1]: https://www.crummy.com/software/BeautifulSoup/bs4/doc.zh/#id30