你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站,请访问 https://docs.azure.cn

适用于 Python 的 Azure 负载测试客户端库 - 版本 1.0.0

Azure 负载测试向用户提供 Python 中的客户端库,用户可以通过该库与 Azure 负载测试服务进行本机交互。 Azure 负载测试是一项完全托管的负载测试服务,可用于生成大规模负载。 该服务可以模拟应用程序的流量,且无需其托管位置。 开发人员、测试人员和质量保证 (QA) 工程师可以使用该服务来优化应用程序性能、可缩放性或容量。

文档

提供了各种文档来帮助你入门

入门

安装包

python -m pip install azure-developer-loadtesting

先决条件

  • 使用此包需要 Python 3.7 或更高版本。
  • 需要 Azure 订阅才能使用此包。
  • 现有的 Azure 开发人员负载测试实例。

使用 Azure Active Directory 凭据创建

若要使用 Azure Active Directory (AAD) 令牌凭据,请提供从 azure 标识 库获取的所需凭据类型的实例。

若要使用 AAD 进行身份验证,必须先安装 pipazure-identity

设置后,可以从 azure.identity 中选择要使用的 凭据 类型。

例如,通过 Azure CLI az login 命令登录, DefaultAzureCredential 将以该用户的身份进行身份验证。

使用返回的令牌凭据对客户端进行身份验证。

创建客户端

Azure 开发人员负载测试 SDK 具有与服务交互的主客户端 (LoadTestingClient) 的 2 个子客户端,即“管理”和“test_run”。

from azure.developer.loadtesting import LoadTestAdministrationClient

# for managing authentication and authorization
# can be installed from pypi, follow: https://pypi.org/project/azure-identity/
# using DefaultAzureCredentials, read more at: https://learn.microsoft.com/en-us/python/api/azure-identity/azure.identity.defaultazurecredential?view=azure-python
from azure.identity import DefaultAzureCredential

client = LoadTestAdministrationClient(endpoint='<endpoint>', credential=DefaultAzureCredential())

<endpoint> 引用资源的数据平面终结点/URL。

关键概念

使用适用于 Python 的 Azure 负载测试客户端库,可以使用客户端与其中每个组件进行交互。 有两个顶级客户端,它们是库的主要入口点

  • LoadTestAdministrationClient (azure.developer.loadtesting.LoadTestAdministrationClient)
  • LoadTestRunClient (azure.developer.loadtesting.LoadTestRunClient)

这两个客户端还具有异步对应项,即

  • LoadTestAdministrationClient (azure.developer.loadtesting.aio.LoadTestAdministrationClient)
  • LoadTestRunClient (azure.developer.loadtesting.aio.LoadTestRunClient)

负载测试管理客户端

LoadTestAdministrationClient用于管理和配置负载测试、应用组件和指标。

测试

测试指定测试脚本和配置设置用于运行负载测试。 可以在 Azure 负载测试资源中创建一个或多个测试。

应用组件

为 Azure 托管应用程序运行负载测试时,可以监视不同 Azure 应用程序组件的资源指标(服务器端指标)。 负载测试运行时以及测试完成后,可以在 Azure 负载测试仪表板中监视和分析资源指标。

指标

在负载测试期间,Azure 负载测试收集有关测试执行的指标。 有两种类型的指标:

  1. 客户端指标会显示测试引擎报告的详细信息。 这些指标包括虚拟用户数、请求响应时间、失败的请求数或每秒请求数。

  2. 服务器端指标适用于 Azure 托管应用程序并显示有关 Azure 应用程序组件的信息。 指标可以是数据库读取次数、HTTP 响应类型或容器资源消耗量。

测试运行客户端

LoadTestRunClient用于启动和停止与负载测试对应的测试运行。 测试运行表示负载测试的一次执行。 它收集与运行 Apache JMeter 脚本关联的日志、负载测试 YAML 配置、要监视的应用组件列表以及测试结果。

Data-Plane 终结点

Azure 负载测试资源的数据平面可使用以下 URL 格式进行寻址:

00000000-0000-0000-0000-000000000000.aaa.cnt-prod.loadtesting.azure.com

第一个 GUID 00000000-0000-0000-0000-000000000000 是用于访问 Azure 负载测试资源的唯一标识符。 接下来 aaa 是资源的 Azure 区域。

数据平面终结点是从控制平面 API 获取的。

示例:1234abcd-12ab-12ab-12ab-123456abcdef.eus.cnt-prod.loadtesting.azure.com

在上面的示例中, eus 表示 Azure 区域 East US

示例

创建负载测试

from azure.developer.loadtesting import LoadTestAdministrationClient
from azure.identity import DefaultAzureCredential
from azure.core.exceptions import HttpResponseError
import os

TEST_ID = "some-test-id"  
DISPLAY_NAME = "my-load-test"  

# set SUBSCRIPTION_ID as an environment variable
SUBSCRIPTION_ID = os.environ["SUBSCRIPTION_ID"]  

client = LoadTestAdministrationClient(endpoint='<endpoint>', credential=DefaultAzureCredential())

try:
    result = client.create_or_update_test(
        TEST_ID,
        {
            "description": "",
            "displayName": "My New Load Test",
            "loadTestConfig": {
                "engineInstances": 1,
                "splitAllCSVs": False,
            },
            "passFailCriteria": {
                "passFailMetrics": {
                    "condition1": {
                        "clientmetric": "response_time_ms",
                        "aggregate": "avg",
                        "condition": ">",
                        "value": 300
                    },
                    "condition2": {
                        "clientmetric": "error",
                        "aggregate": "percentage",
                        "condition": ">",
                        "value": 50
                    },
                    "condition3": {
                        "clientmetric": "latency",
                        "aggregate": "avg",
                        "condition": ">",
                        "value": 200,
                        "requestName": "GetCustomerDetails"
                    }
                }
            },
            "secrets": {
                "secret1": {
                    "value": "https://sdk-testing-keyvault.vault.azure.net/secrets/sdk-secret",
                    "type": "AKV_SECRET_URI"
                }
            },
            "environmentVariables": {
                "my-variable": "value"
            }
        }
    )
    print(result)
except HttpResponseError as e:
     print('Service responded with error: {}'.format(e.response.json()))

将 .jmx 文件上传到测试

from azure.developer.loadtesting import LoadTestAdministrationClient
from azure.identity import DefaultAzureCredential
from azure.core.exceptions import HttpResponseError

TEST_ID = "some-test-id"  
FILE_NAME = "some-file-name.jmx"  

client = LoadTestAdministrationClient(endpoint='<endpoint>', credential=DefaultAzureCredential())

try:

    # uploading .jmx file to a test
    resultPoller = client.begin_upload_test_file(TEST_ID, FILE_NAME, open("sample.jmx", "rb"))

    # getting result of LRO poller with timeout of 600 secs
    validationResponse = resultPoller.result(600)
    print(validationResponse)
    
except HttpResponseError as e:
    print("Failed with error: {}".format(e.response.json()))

运行测试

from azure.developer.loadtesting import LoadTestRunClient
from azure.identity import DefaultAzureCredential
from azure.core.exceptions import HttpResponseError

TEST_ID = "some-test-id"  
TEST_RUN_ID = "some-testrun-id" 
DISPLAY_NAME = "my-load-test-run"  

client = LoadTestRunClient(endpoint='<endpoint>', credential=DefaultAzureCredential())

try:
    testRunPoller = client.begin_test_run(
    TEST_RUN_ID,
        {
            "testId": TEST_ID,
            "displayName": "My New Load Test Run",
        }
    )

    #waiting for test run status to be completed with timeout = 3600 seconds
    result = testRunPoller.result(3600)
    
    print(result)
except HttpResponseError as e:
    print("Failed with error: {}".format(e.response.json()))

后续步骤

在此处找到更多示例。

贡献

本项目欢迎贡献和建议。 大多数贡献要求你同意贡献者许可协议 (CLA),并声明你有权(并且确实有权)授予我们使用你的贡献的权利。 有关详细信息,请访问 https://cla.microsoft.com

提交拉取请求时,CLA 机器人将自动确定你是否需要提供 CLA,并相应地修饰 PR(例如标签、注释)。 直接按机器人提供的说明操作。 只需使用 CLA 对所有存储库执行一次这样的操作。

本项目采用 Microsoft 开源行为准则。 有关详细信息,请参阅“行为准则常见问题解答”,如有其他任何问题或意见,请联系 opencode@microsoft.com。

故障排除

即将推出更多相关信息...