怎么用Python+Pytest實現(xiàn)壓力測試

這篇“怎么用Python+Pytest實現(xiàn)壓力測試”文章的知識點大部分人都不太理解，所以小編給大家總結(jié)了以下內(nèi)容，內(nèi)容詳細，步驟清晰，具有一定的借鑒價值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來看看這篇“怎么用Python+Pytest實現(xiàn)壓力測試”文章吧。

1.程序說明

1.1 設(shè)置測試參數(shù)

首先做的第一件事情就是設(shè)置測試參數(shù)。代碼如下

# 定義測試用例  
def test_performance():  
    # 設(shè)置測試參數(shù)  
    url = 'http://www.a.com/'  
    num_threads = 20  
    num_requests = 200  
    timeout = 5

這里面設(shè)置了網(wǎng)站的URL， 線程數(shù)， 每個線程的請求次數(shù)，以及超時時間。 可以看到， 這里面一共會做4000次請求。

1.2 初始化測試結(jié)果

這里做一個提示：注意縮進， 這段代碼仍然在測試用例test_performance內(nèi)。

    # 初始化測試結(jié)果  
    response_times = []  
    errors = 0  
    successes = 0

1.3 定義測試函數(shù)

接下來， 定義一個內(nèi)部函數(shù)。這個函數(shù)就是在某一線程內(nèi)完成設(shè)定次數(shù)的請求。

    # 定義測試函數(shù)  
    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 創(chuàng)建線程、執(zhí)行線程、等待

    # 創(chuàng)建測試線程  
    threads = []  
    for _ in range(num_threads):  
        t = threading.Thread(target=test_func)  
        threads.append(t)  
      
    # 啟動測試線程  
    for t in threads:  
        t.start()  
      
    # 等待測試線程結(jié)束  
    for t in threads:  
        t.join()

1.5 計算測試結(jié)果

    # 計算測試結(jié)果  
    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 將測試結(jié)果寫入文件

    # 將測試結(jié)果寫入文件  
    with open('performance_test_result.txt', 'w') as f:  
        f.write(f'總請求數(shù)：{total_requests}\n')  
        f.write(f'總時間：{sum(response_times):.2f}s\n')  
        f.write(f'吞吐量：{throughput:.2f} requests/s\n')  
        f.write(f'并發(fā)數(shù)：{concurrency}\n')  
        f.write(f'錯誤率：{error_rate:.2%}\n')  
        f.write(f'CPU利用率：{cpu_usage:.2f}%\n')  
        f.write(f'內(nèi)存利用率：{memory_usage:.2f}%\n')

2.程序執(zhí)行

2.1 直接執(zhí)行

在PyCharm里面直接執(zhí)行這段代碼， 得出的結(jié)果是：

總請求數(shù)：4000  
總時間：1837.65s  
吞吐量：2.17 requests/s  
并發(fā)數(shù)：20  
錯誤率：0.12%  
CPU利用率：4.10%  
內(nèi)存利用率：88.60%

2.2 加個裝飾器然后出報告

如果在PyCharm里面直接執(zhí)行上面的代碼， 雖然我們把結(jié)果寫在文件中，但是， 不好看呀。

所以呢，再額外介紹一個方法，這個方法能夠生成一個相對美觀的測試報告出來。

2.2.1 聲明壓力測試

首先在定義用例的時候通過裝飾器聲明這是一個壓力測試：

# 定義測試用例  
@pytest.mark.performance  
def test_performance():  
    # 設(shè)置測試參數(shù)  
    url = 'http://www.a.biz/'  
    num_threads = 20

2.2.2 在命令行中通過pytest命令執(zhí)行測試

第二步， 在命令行中執(zhí)行測試

• -v 用于顯示詳細的測試結(jié)果

• --html 用于指定輸出報告的位置。 這個參數(shù)需要依賴包：pytest-html

$ pytest  -v --html=report.html  test_a.py   

輸出執(zhí)行結(jié)果是：

======================== 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.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-grp\\jdk\\', 'Base URL': ''}
rootdir: E:\develop\python\pytest-training\test
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:\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>

最終生成的報告是：（有點長， 截取了關(guān)鍵部分）

3.案例缺陷

因為時間關(guān)系， 本案例今天沒有時間在服務(wù)器端執(zhí)行， 所以通過psutil庫所取得CPU利用率和內(nèi)存利用率時間并不對。 如果是在服務(wù)器端執(zhí)行， 這兩個數(shù)字才是對的。

如果要在本地獲取服務(wù)器的CPU，內(nèi)存，IO等情況，有一個監(jiān)控神器：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():  
    # 設(shè)置測試參數(shù)  
    url = 'http://www.tdouya.biz/'  
    num_threads = 20  
    num_requests = 200  
    timeout = 5  
  
    # 初始化測試結(jié)果  
    response_times = []  
    errors = 0  
    successes = 0  
  
    # 定義測試函數(shù)  
    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  
  
    # 創(chuàng)建測試線程  
    threads = []  
    for _ in range(num_threads):  
        t = threading.Thread(target=test_func)  
        threads.append(t)  
  
    # 啟動測試線程  
    for t in threads:  
        t.start()  
  
    # 等待測試線程結(jié)束  
    for t in threads:  
        t.join()  
  
    # 計算測試結(jié)果  
    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  
  
    # 輸出測試結(jié)果  
    print(f'總請求數(shù)：{total_requests}')  
    print(f'總時間：{sum(response_times):.2f}s')  
    print(f'吞吐量：{throughput:.2f} requests/s')  
    print(f'并發(fā)數(shù)：{concurrency}')  
    print(f'錯誤率：{error_rate:.2%}')  
    print(f'CPU利用率：{cpu_usage:.2f}%')  
    print(f'內(nèi)存利用率：{memory_usage:.2f}%')  
  
    # 將測試結(jié)果寫入文件  
    with open('performance_test_result.txt', 'w') as f:  
        f.write(f'總請求數(shù)：{total_requests}\n')  
        f.write(f'總時間：{sum(response_times):.2f}s\n')  
        f.write(f'吞吐量：{throughput:.2f} requests/s\n')  
        f.write(f'并發(fā)數(shù)：{concurrency}\n')  
        f.write(f'錯誤率：{error_rate:.2%}\n')  
        f.write(f'CPU利用率：{cpu_usage:.2f}%\n')  
        f.write(f'內(nèi)存利用率：{memory_usage:.2f}%\n')

以上就是關(guān)于“怎么用Python+Pytest實現(xiàn)壓力測試”這篇文章的內(nèi)容，相信大家都有了一定的了解，希望小編分享的內(nèi)容對大家有幫助，若想了解更多相關(guān)的知識內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道。