Python+Pytest实现压力测试详解

本是古典何须时尚 2023-10-16 13:25 93阅读 0赞

![3fb423f2336049aa8081306659ced249.png][] 在现代Web应用程序中，性能是至关重要的。为了确保应用程序能够在高负载下正常运行，我们需要进行性能测试。本文就来用Pytest进行压力测试，希望对大家有所帮助

−

###### 目录 ######

*  1.程序说明
    
     *  1.1 设置测试参数
     *  1.2 初始化测试结果
     *  1.3 定义测试函数
     *  1.4 创建线程、执行线程、等待
     *  1.5 计算测试结果
     *  1.6 将测试结果写入文件
 *  2.程序执行
    
     *  2.1 直接执行
     *  2.2 加个装饰器然后出报告
 *  3.案例缺陷
 *  4 完整源码

在现代Web应用程序中，性能是至关重要的。为了确保应用程序能够在高负载下正常运行，我们需要进行性能测试。 今天，应小伙伴的提问， 田辛老师来写一个Pytest进行压力测试的简单案例。 这个案例的测试网站我们就隐藏了，不过网站的基本情况是：

*  阿里
 *  框架：FastAdmin.net

### 1.程序说明 ###

#### 1.1 设置测试参数 ####

首先，做的第一件事情就是设置测试参数。代码如下

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td> <p></p> <p># 定义测试用例</p> <p>def test_performance():</p> <p># 设置测试参数</p> <p>url = 'http://www.a.com/'</p> <p>num_threads = 20</p> <p>num_requests = 200</p> <p>timeout = 5</p> <p></p> </td> 
  </tr> 
 </tbody> 
</table>

设置网站的URL， 线程数， 每个线程的请求次数，以及超时时间。 可以看到一共会做4000次请求。

#### 1.2 初始化测试结果 ####

这段代码， 这里做一个提示：注意缩进， 这段代码仍然在测试用例test\_performance内。

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td> <p></p> <p>&nbsp; &nbsp; # 初始化测试结果 &nbsp;</p> <p>&nbsp; &nbsp; response_times = [] &nbsp;</p> <p>&nbsp; &nbsp; errors = 0 &nbsp;</p> <p>&nbsp; &nbsp; successes = 0</p> </td> 
  </tr> 
 </tbody> 
</table>

### 1.3 定义测试函数 ###

接下来， 定义一个内部函数。这个函数就是在某一线程内完成设定次数的请求。

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td> <p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;# 定义测试函数 &nbsp; &nbsp; &nbsp;def test_func(): &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;nonlocal errors, successes &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;for _ in range(num_requests): &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;try: &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;start_time = time.time() &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;requests.get(url, timeout=timeout) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;end_time = time.time() &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;response_time = end_time - start_time &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;response_times.append(response_time) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;successes += 1 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;except requests.exceptions.RequestException: &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;errors += 1</p> </td> 
  </tr> 
 </tbody> 
</table>

### 1.4 创建线程、执行线程、等待 ###

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td> <p></p> <p>&nbsp; &nbsp; # 创建测试线程 &nbsp;</p> <p>&nbsp; &nbsp; threads = [] &nbsp;</p> <p>&nbsp; &nbsp; for _ in range(num_threads): &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; t = threading.Thread(target=test_func) &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; threads.append(t) &nbsp;</p> <p>&nbsp; &nbsp; &nbsp;&nbsp;</p> <p>&nbsp; &nbsp; # 启动测试线程 &nbsp;</p> <p>&nbsp; &nbsp; for t in threads: &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; t.start() &nbsp;</p> <p>&nbsp; &nbsp; &nbsp;&nbsp;</p> <p>&nbsp; &nbsp; # 等待测试线程结束 &nbsp;</p> <p>&nbsp; &nbsp; for t in threads: &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; t.join()</p> </td> 
  </tr> 
 </tbody> 
</table>

### 1.5 计算测试结果 ###

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td> <p>&nbsp;# 计算测试结果 &nbsp;</p> <p>&nbsp; &nbsp; total_requests = num_threads * num_requests &nbsp;</p> <p>&nbsp; &nbsp; throughput = successes / (sum(response_times) or 1) &nbsp;</p> <p>&nbsp; &nbsp; concurrency = num_threads &nbsp;</p> <p>&nbsp; &nbsp; error_rate = errors / (total_requests or 1) &nbsp;</p> <p>&nbsp; &nbsp; cpu_usage = psutil.cpu_percent() &nbsp;</p> <p>&nbsp; &nbsp; memory_usage = psutil.virtual_memory().percent</p> </td> 
  </tr> 
 </tbody> 
</table>

#### 1.6 将测试结果写入文件 ####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td> <p># 将测试结果写入文件 &nbsp;</p> <p>&nbsp; &nbsp; with open('performance_test_result.txt', 'w') as f: &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; f.write(f'总请求数：{total_requests}\n') &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; f.write(f'总时间：{sum(response_times):.2f}s\n') &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; f.write(f'吞吐量：{throughput:.2f} requests/s\n') &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; f.write(f'并发数：{concurrency}\n') &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; f.write(f'错误率：{error_rate:.2%}\n') &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; f.write(f'CPU利用率：{cpu_usage:.2f}%\n') &nbsp;</p> <p>&nbsp; &nbsp; &nbsp; &nbsp; f.write(f'内存利用率：{memory_usage:.2f}%\n')</p> </td> 
  </tr> 
 </tbody> 
</table>

### 2.程序执行 ###

#### 2.1 直接执行 ####

在PyCharm里面直接执行这段代码， 得出的结果是：

> 总请求数：4000 总时间：1837.65s 吞吐量：2.17 requests/s 并发数：20 错误率：0.12% CPU利用率：4.10% 内存利用率：88.60%

#### 2.2 加个装饰器然后出报告 ####

如果在PyCharm里面直接执行上面的代码， 虽然我们把结果写在文件中，但是， 不好看呀。

额外介绍一个方法，这个方法能够生成一个相对美观的测试报告出来。

**2.2.1 声明压力测试**

首先在定义用例的时候通过装饰器声明这是一个压力测试：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td> <p># 定义测试用例 &nbsp;</p> <p>@pytest.mark.performance &nbsp;</p> <p>def test_performance(): &nbsp;</p> <p>&nbsp; &nbsp; # 设置测试参数 &nbsp;</p> <p>&nbsp; &nbsp; url = 'http://www.a.biz/' &nbsp;</p> <p>&nbsp; &nbsp; num_threads = 20</p> </td> 
  </tr> 
 </tbody> 
</table>

**2.2.2 在命令行中通过pytest命令执行测试**

第二步， 在命令行中执行测试

*  \-v 用于显示详细的测试结果
 *  \--html 用于指定输出报告的位置。 这个参数需要依赖包：pytest-html

> $ pytest -v --html=report.html test\_a.py

输出执行结果是：

> ======================== test session starts =================================platform win32 -- Python 3.10.9, pytest-7.2.1, pluggy-1.0.0 -- D:\\python-grp\\miniconda\_env\\py3.10\_playwright\\python.execachedir: .pytest\_cachemetadata: \{'Python': '3.10.9', 'Platform': 'Windows-10-10.0.22624-SP0', 'Packages': \{'pytest': '7.2.1', 'pluggy': '1.0.0'\}, 'Plugins': \{'allure-pytest': '2.12.0', 'base-url': '2.0.0', 'html': '3.2.0', 'metadata': '2.0.4', 'ordering': '0.6', 'playwright': '0.3.0'\}, 'JAVA\_HOME': 'D:\\\\java-grp\\\\jdk\\\\', 'Base URL': ''\}rootdir: E:\\develop\\python\\pytest-training\\testplugins: allure-pytest-2.12.0, base-url-2.0.0, html-3.2.0, metadata-2.0.4, ordering-0.6, playwright-0.3.0collected 1 item test\_a.py::test\_performance PASSED \[100%\]========================== warnings summary ================================= test\_a.py:25 E:\\develop\\python\\pytest-training\\test\\test\_a.py:25: PytestUnknownMarkWarning: Unknown pytest.mark.performance - is this a typo? You can register custom marks to avoid this warning - for details, see https://docs.pytest.org/en/stable/how-to/mark.html @pytest.mark.performance-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html-- generated html file: file:///E:/develop/python/pytest-training/test/report.html -- ================= 1 passed, 1 warning in 99.09s (0:01:39) =================== (D:\\python-grp\\miniconda\_env\\py3.10\_playwright) E:\\develop\\python\\pytest-training\\test>

最终生成的报告是：（截取了关键部分）

![图片][de30e347cb8b1ea90325c53edbad7cd8.jpeg]

### 3.案例缺陷 ###

因为时间关系， 本案例今天没有时间在服务器端执行， 所以通过psutil库所取得CPU利用率和内存利用率时间并不对。 如果是在服务器端执行， 这两个数字才是对的。

如果要在本地获取服务器的CPU，内存，IO等情况，有一个监控神器：Prometheus。不过这东西配置起来又是另一个话题。

### 4 完整源码 ###

# 压力测试案例  
    
    
    
    import threading  
    
    import time  
    
      
    
    import psutil  
    
    import pytest  
    
    import requests  
    
      
    
      
    
    # 定义测试用例  
    
    @pytest.mark.performance  
    
    def test_performance():  
    
        # 设置测试参数  
    
        url = 'http://www.tdouya.biz/'  
    
        num_threads = 20  
    
        num_requests = 200  
    
        timeout = 5  
    
      
    
        # 初始化测试结果  
    
        response_times = []  
    
        errors = 0  
    
        successes = 0  
    
      
    
        # 定义测试函数  
    
        def test_func():  
    
            nonlocal errors, successes  
    
            for _ in range(num_requests):  
    
                try:  
    
                    start_time = time.time()  
    
                    requests.get(url, timeout=timeout)  
    
                    end_time = time.time()  
    
                    response_time = end_time - start_time  
    
                    response_times.append(response_time)  
    
                    successes += 1  
    
                except requests.exceptions.RequestException:  
    
                    errors += 1  
    
      
    
        # 创建测试线程  
    
        threads = []  
    
        for _ in range(num_threads):  
    
            t = threading.Thread(target=test_func)  
    
            threads.append(t)  
    
      
    
        # 启动测试线程  
    
        for t in threads:  
    
            t.start()  
    
      
    
        # 等待测试线程结束  
    
        for t in threads:  
    
            t.join()  
    
      
    
        # 计算测试结果  
    
        total_requests = num_threads * num_requests  
    
        throughput = successes / (sum(response_times) or 1)  
    
        concurrency = num_threads  
    
        error_rate = errors / (total_requests or 1)  
    
        cpu_usage = psutil.cpu_percent()  
    
        memory_usage = psutil.virtual_memory().percent  
    
      
    
        # 输出测试结果  
    
        print(f'总请求数：{total_requests}')  
    
        print(f'总时间：{sum(response_times):.2f}s')  
    
        print(f'吞吐量：{throughput:.2f} requests/s')  
    
        print(f'并发数：{concurrency}')  
    
        print(f'错误率：{error_rate:.2%}')  
    
        print(f'CPU利用率：{cpu_usage:.2f}%')  
    
        print(f'内存利用率：{memory_usage:.2f}%')  
    
      
    
        # 将测试结果写入文件  
    
        with open('performance_test_result.txt', 'w') as f:  
    
            f.write(f'总请求数：{total_requests}\n')  
    
            f.write(f'总时间：{sum(response_times):.2f}s\n')  
    
            f.write(f'吞吐量：{throughput:.2f} requests/s\n')  
    
            f.write(f'并发数：{concurrency}\n')  
    
            f.write(f'错误率：{error_rate:.2%}\n')  
    
            f.write(f'CPU利用率：{cpu_usage:.2f}%\n')  
    
            f.write(f'内存利用率：{memory_usage:.2f}%\n')

到此这篇关于Python+Pytest实现压力测试详解的文章就介绍到这了。

**最后：**下方这份完整的软件测试视频教程已经整理上传完成，需要的朋友们可以自行领取**【保证100%免费】**

![8b0171187dce405e878f274a8cdd9ec0.png][]

#### **软件测试面试文档** ####

我们学习必然是为了找到高薪的工作，下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料，并且有字节大佬给出了权威的解答，刷完这一套面试资料相信大家都能找到满意的工作。

![a3e4226c935c4242804cbdf26b571bc6.png][]

![58df6cd288094ad8874e814f2477620b.gif][]

[3fb423f2336049aa8081306659ced249.png]: https://img-blog.csdnimg.cn/3fb423f2336049aa8081306659ced249.png
[de30e347cb8b1ea90325c53edbad7cd8.jpeg]: https://img-blog.csdnimg.cn/img_convert/de30e347cb8b1ea90325c53edbad7cd8.jpeg
[8b0171187dce405e878f274a8cdd9ec0.png]: https://img-blog.csdnimg.cn/8b0171187dce405e878f274a8cdd9ec0.png
[a3e4226c935c4242804cbdf26b571bc6.png]: https://img-blog.csdnimg.cn/a3e4226c935c4242804cbdf26b571bc6.png
[58df6cd288094ad8874e814f2477620b.gif]: https://img-blog.csdnimg.cn/58df6cd288094ad8874e814f2477620b.gif