# 定義測試用例  
def test_performance():  
    # 設(shè)置測試參數(shù)  
    url = 'http://www.a.com/'  
    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é)果寫入文件  
    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')

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

#!/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')