使用locust in python:介绍、关联变量和基本断言

Locust 是一个很好的开源负载测试工具,可以为有 Python 经验的开发人员提供测试,因为测试可以作为代码创建。我们在之前的一些文章中已经讨论过这个问题。但是如果你从来没有用过它呢?在这篇文章中,我们将向你展示如何使用这个性能测试工具开始你的第一步,通过展示一个用 Python 开发的基本工作流程的例子,一个高级的工作流程,以及如何关联贵重物品和断言你的脚。

开始: 安装 Python

为了运行 Locust,你需要安装 Python。如果这不是你的情况,我们将留给你一个链接下载它在这里。然后,你所要做的就是运行以下命令:

pip3 install locust

从 Locust GUI 脚本和执行负载测试

下一步是为名为 locustfile.py 的脚本创建一个文件。在这个文件中,我们将定义在负载测试中执行的 HTTP 请求。使用该脚本的文件名使 Locust 能够自动找到该文件。(如果要为文件使用另一个名称,在执行时需要添加参数 -f 和文件名。我将在下面向你展示。

在这个例子中,我们将加载测试站点的 https://www.demoblaze.com/。

创建一个基本脚本

首先,我们将编写一个脚本,调用 demoblaze 主页,如下所示:

from locust import HttpUser, task
            
class User(HttpUser):
    @task
    def mainPage(self):
        self.client.get("/")

注意,脚本中没有指定被测试站点的 url。相反,它是在运行测试时从 UI 指定的。 Locust 将只在 decorator@任务中运行函数,因此我们必须记住添加它。对于定义了多个任务的情况,默认情况下它们将随机执行。

Running the Script 2. 运行脚本

为了运行这个测试,我们需要从命令行执行脚本目录中的 locust 命令,这将启动端口8089上的 web 用户界面。只需在浏览器上导航到 http://localhost:8089地图就可以访问它。

注意: 如果您在另一个程序使用端口时得到一个错误,您可以使用 locust 命令 -- web-port [ port ]更改它。正如我上面提到的,如果您想运行一个不同名称的脚本,可以执行 locust-f [ file name ]。

用户的数量,每秒启动的用户数量和主机(测试中的 URL)可以从 Locust 的 web 用户界面中选择。Locust 将使用脚本中的路径和这里指定的主机为每个请求创建完整的 url。

现在你可以通过点击 Start swarming 按钮来运行测试,浏览器会显示一个如下图所示的仪表盘:

在这个仪表板中,我们将看到一个报告,其中包括执行的请求数量、失败请求数量、90% 以及其他实时统计信息。默认情况下,虚拟用户将继续运行,直到测试停止。

如果你不想使用 web UI 来运行脚本,你可以运行以下命令:

locust --headless -u 1 -r 1 -H https://www.demoblaze.com

其中 -u 指定用户数,-r 表示产生率,-h 表示主机。

创建更高级的脚本

现在我们了解了如何创建和运行基本测试,接下来让我们用更复杂的工作流来做一个测试。

这个工作流程包括4个步骤:

进入网站 登录 向购物车中添加产品 提交购物车

一旦我们获得了工作流中每个步骤的 HTTP 请求,我们就可以为每个用户操作创建一个方法。

from locust import HttpUser, SequentialTaskSet, task, between
            
class User(HttpUser):    
    @task
    class SequenceOfTasks(SequentialTaskSet):
        wait_time = between(1, 5)
        @task
        def mainPage(self):
            self.client.get("/")
            self.client.get("https://api.demoblaze.com/entries")
        @task
        def login(self):
            self.client.options("https://api.demoblaze.com/login")
            self.client.post("https://api.demoblaze.com/login",json={"username":"aaaa","password":"YWFhYQ=="})
            self.client.options("https://api.demoblaze.com/check")
            self.client.get("https://api.demoblaze.com/entries")
            self.client.post("https://api.demoblaze.com/check",json={"token":"YWFhYTE2MzA5NDU="})            
        @task
        def clickProduct(self):
            self.client.get("/prod.html?idp_=1")
            self.client.options("https://api.demoblaze.com/check")
            self.client.options("https://api.demoblaze.com/view")
            self.client.post("https://api.demoblaze.com/check",json={"token":"YWFhYTE2MzA5NDU="})
            self.client.post("https://api.demoblaze.com/view",json={"id":"1"})
        @task
        def addToCart(self):
            self.client.options("https://api.demoblaze.com/addtocart")
            self.client.post("https://api.demoblaze.com/addtocart",json={"id":"fb3d5d23-f88c-80d9-a8de-32f1b6034bfd","cookie":"YWFhYTE2MzA5NDU=","prod_id":1,"flag":'true'})
        @task 
        def viewCart(self):
            self.client.get("/cart.html")
            self.client.options("https://api.demoblaze.com/check")
            self.client.options("https://api.demoblaze.com/viewcart")
            self.client.post("https://api.demoblaze.com/check",json={"token":"YWFhYTE2MzA5NDU="})
            self.client.post("https://api.demoblaze.com/viewcart",json={"cookie":"YWFhYTE2MzA5NDU=","flag":'true'})
            self.client.options("https://api.demoblaze.com/view")
            self.client.post("https://api.demoblaze.com/check",json={"token":"YWFhYTE2MzA5NDU="})
            self.client.post("https://api.demoblaze.com/view",json={"id":"1"})

正如您所看到的,所有方法都使用 SequentialTaskSet 包含在类中,因此它们可以按照声明的相同顺序执行。

另外,通过使用 wait _ time,我们可以在任务之间添加一个暂停。在这种情况下,是1到5秒之间的随机停顿。

您还会注意到,对 API 的请求是使用完整的 URL 编写的,因为 Locust web 用户界面只允许在 host 字段中使用一个 URL。

相关变量

下一步是将硬编码令牌关联起来。关联动态的参数非常重要,我们知道这个标记在用户每次登录时都会发生变化。令牌可能过期,如果没有将其参数化,脚本就会停止工作。

通过分析 HTTP 请求工作流,我们可以看到可以从/login 响应中提取/check post 中发送的令牌。然后,我们可以将它保存到一个变量中,并在以下所有需要它的请求中使用它。

响应的标记格式如下:

"Auth_token: YWFhYTE2MzA1ODg=" 因此可以使用下面的正则表达式来提取它:

"Auth_token: (.+?)" 接下来,需要导入模块 re,并使用 match 方法将提取的值保存到变量中。

因此,为了从响应中提取标记,我们首先保存对变量的响应,如下所示:

response = self.client.post("https://api.demoblaze.com/login",json={"username":"aaaa","password":"YWFhYQ=="})
现在我们可以定义一个全局变量并使用正则表达式提取标记。

global token 
  token = re.match("\"Auth_token: (.+?)\"",response.text)[1]

我们可以在以下请求中使用该变量

self.client.post("https://api.demoblaze.com/check",json={"token":token} 这就是登录和点击产品交易的样子:

@task
        def login(self):
            self.client.options("https://api.demoblaze.com/login")
            response = self.client.post("https://api.demoblaze.com/login",json={"username":"aaaa","password":"YWFhYQ=="})
            global token 
            token = re.match("\"Auth_token: (.+?)\"",response.text)[1]
            self.client.options("https://api.demoblaze.com/check")
            self.client.get("https://api.demoblaze.com/entries")
            self.client.post("https://api.demoblaze.com/check",json={"token":token})            
        @task
        def clickProduct(self):
            self.client.get("/prod.html?idp_=1")
            self.client.options("https://api.demoblaze.com/check")
            self.client.options("https://api.demoblaze.com/view")
            self.client.post("https://api.demoblaze.com/check",json={"token":token})
            self.client.post("https://api.demoblaze.com/view",json={"id":"1"})

注意: 在这个例子中,我们只使用了一个用户。如果你想学习如何在多个用户中运行脚本,你可以查看这篇文章。

https://www.blazemeter.com/blog/how-to-run-locust-with-different-users

使用断言

现在,我将向您展示如何添加一个简单的断言来验证添加到购物车中的产品是否被正确添加。

Locust 没有很多内置功能,但是使用 Python 可以轻松地添加自定义功能。

Failure (“ Error message”)可用于将请求标记为 failed。为了让断言正常工作,这个函数应该在 if 子句中使用,并且应该添加 catch _ response 参数来验证响应,如下所示。

@task 
        def viewCart(self):
            self.client.get("/cart.html")
            self.client.options("https://api.demoblaze.com/check")
            self.client.options("https://api.demoblaze.com/viewcart")
            self.client.post("https://api.demoblaze.com/check",json={"token":token})
            with self.client.post("https://api.demoblaze.com/viewcart",catch_response=True,json={"cookie":token,"flag":'true'}) as response:
                if '"prod_id":1' not in response.text:
                    response.failure("Assert failure, response does not contain expected prod_id")
            self.client.options("https://api.demoblaze.com/view")
            self.client.post("https://api.demoblaze.com/check",json={"token":token})
            self.client.post("https://api.demoblaze.com/view",json={"id":"1"})
            

如果我们运行测试,我们可以看到它没有错误。但是,我们如何知道这种主张是否有效呢?让我们将产品 id 更改为响应中不存在的 id。

with self.client.post("https://api.demoblaze.com/viewcart",catch_response=True,json={"cookie":token,"flag":'true'}) as response:
                if '"prod_id":1234' not in response.text:
                    response.failure("Assert failure, response does not contain expected prod_id")

当再次执行测试时,/viewcart POST 将失败,并且定义的错误消息可以在 Failures 选项卡中看到。

如果你要使用多个断言,最好创建一个函数,避免重写类似的代码,比如这个:

def assertContains(response,text):
    with response as r:
        if text not in r.text:
            r.failure("Expected "+ response.text + " to contain "+ text)

你可以像下面这样调用它,每次你想用文本做一个断言:

            assertContains(self.client.post("https://api.demoblaze.com/viewcart",catch_response=True,json={"cookie":token,"flag":'true'}),'"prod_id":1')

一旦你完成locust脚本,你可以运行它在 BlazeMeter 的规模,集成在 CI/CD 和看到先进的报告。现在就开始。

author

石头 磊哥 seven 随便叫

company

HSBC 大家好,我已经加入了HSBC

roles

QA(营生) dev(front-end dev 兴趣爱好)

联系方式

如果想转载或者高薪挖我 请直接联系我 哈哈

wechat:

qileiwangnan

email:

qileilove@gmail.com