«

怎么用Python+Pytest实现压力测试

时间:2024-8-2 10:48     作者:韩俊     分类: Python


这篇“怎么用Python+Pytest实现压力测试”文章的知识点大部分人都不太理解,所以小编给大家总结了以下内容,内容详细,步骤清晰,具有一定的借鉴价值,希望大家阅读完这篇文章能有所收获,下面我们一起来看看这篇“怎么用Python+Pytest实现压力测试”文章吧。

    1.程序说明

    1.1 设置测试参数

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

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

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

    1.2 初始化测试结果

    这里做一个提示:注意缩进, 这段代码仍然在测试用例test_performance内。

        # 初始化测试结果  
        response_times = []  
        errors = 0  
        successes = 0

    1.3 定义测试函数

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

        # 定义测试函数  
        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

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

        # 创建测试线程  
        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()

    1.5 计算测试结果

        # 计算测试结果  
        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

    1.6 将测试结果写入文件

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

    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 声明压力测试

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

    # 定义测试用例  
    @pytest.mark.performance  
    def test_performance():  
        # 设置测试参数  
        url = 'http://www.a.biz/'  
        num_threads = 20

    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-grpminiconda_envpy3.10_playwrightpython.exe
    cachedir: .pytest_cache
    metadata: {'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-grpjdk', 'Base URL': ''}
    rootdir: E:developpythonpytest-training est
    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
    collected 1 item                                                                                                                                                                 

    test_a.py::test_performance PASSED                                                                                                                                 [100%]

    ========================== warnings summary ================================= 
    test_a.py:25
      E:developpythonpytest-training est est_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-grpminiconda_envpy3.10_playwright) E:developpythonpytest-training est>

    最终生成的报告是:(有点长, 截取了关键部分)

    3.案例缺陷

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

    如果要在本地获取服务器的CPU,内存,IO等情况,有一个监控神器:Prometheus。

    4 完整源码

    #!/usr/bin/env python  
    # -*- coding:utf-8 -*-  
    """  
    #-----------------------------------------------------------------------------  
    #                     --- TDOUYA STUDIOS ---  
    #-----------------------------------------------------------------------------  
    #  
    # @Project : pytest-training  
    # @File    : test_a.py  
    # @Author  : tianxin.xp@gmail.com  
    # @Date    : 2023/3/10 14:39  
    #  
    # 压力测试案例  
    #  
    #--------------------------------------------------------------------------"""  
    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}
    ')  
            f.write(f'总时间:{sum(response_times):.2f}s
    ')  
            f.write(f'吞吐量:{throughput:.2f} requests/s
    ')  
            f.write(f'并发数:{concurrency}
    ')  
            f.write(f'错误率:{error_rate:.2%}
    ')  
            f.write(f'CPU利用率:{cpu_usage:.2f}%
    ')  
            f.write(f'内存利用率:{memory_usage:.2f}%
    ')

    标签: python

    热门推荐