性能测试 - LocustIO 验证请求结果

Max.Bai

2017-04

1. 默认验证请求结果

Locust 默认会对请求做校验，如果是200就是成功的，不然就是失败

不用做任何操作，在task里面直接请求就可以了, locust会记录请求的结果并加以统计

self.client.get('/getuser', param)

2. 自定义验证方式

大部分时候我们需要自己校验请求结果，比如我们不只校验http状态，我们还需要校验请求内容，或者说我们认为404 是成功的，总之就是默认的校验方式无法满足我们的需要。

举个栗子：

我们需要校验http状态同时还需要校验返回内容（json）里面的code值为1000

那么我们需要做的时：

1. 当然自己校验的话请求里面就要带上非常重要的参数了（catch_response=True）

rsp = self.client.post(url, data=payload, headers=headers, catch_response=True, cookies=cookies)

2. http 状态校验，必须的, 自己校验后locust就不做任何校验了

response.raise_for_status()

3. http校验成功的情况下，业务数据校验，当然根据你的业务来了

code = response.json().get("code")
if code == 1000:
    print("success")
else:
    print("errrrrrrrrror")

4. 也非常重要，自己校验locust就不做任何校验，那么同时请求事件也不会做记录，应为他不知道请求是成功还是失败，他不知道你的标准，那同时自己要写校验事件进去，不然web页面就没有请求数据和结果数据了
用到的方法就是

# for assert faile
events.request_failure.fire(
                    request_type=response.locust_request_meta["method"],
                    name=response.locust_request_meta["name"],
                    response_time=response.locust_request_meta["response_time"],
                    response_length=response.locust_request_meta["content_size"],
                    exception="Response Code Error! Code:{0}".format(code)
                )

# for assert success
events.request_success.fire(
                    request_type=response.locust_request_meta["method"],
                    name=response.locust_request_meta["name"],
                    response_time=response.locust_request_meta["response_time"],
                    response_length=response.locust_request_meta["content_size"],
                )

3. Demo完整代码

# -*- coding:utf-8 -*-


import json
from locust import HttpLocust, TaskSet, task, events
from requests.exceptions import (RequestException, MissingSchema,
    InvalidSchema, InvalidURL)

class WebsiteTasks(TaskSet):
    
    def on_start(self):
        pass

    def check_response(self, response):
        try:
            response.raise_for_status()
        except RequestException as e:
            events.request_failure.fire(
                request_type=response.locust_request_meta["method"],
                name=response.locust_request_meta["name"],
                response_time=response.locust_request_meta["response_time"],
                exception=e,
            )
        else:
            code = response.json().get("code")
            if code == 1000:
                events.request_success.fire(
                    request_type=response.locust_request_meta["method"],
                    name=response.locust_request_meta["name"],
                    response_time=response.locust_request_meta["response_time"],
                    response_length=response.locust_request_meta["content_size"],
                )
            else:
                events.request_failure.fire(
                    request_type=response.locust_request_meta["method"],
                    name=response.locust_request_meta["name"],
                    response_time=response.locust_request_meta["response_time"],
                    response_length=response.locust_request_meta["content_size"],
                    exception="Response Code Error! Code:{0}".format(code)
                )
    @task
    def requests_getChatMessage(self):
        cookies = {"_Qt":"1490868798569",
            "_QM":"1",
            "LoginPc_newType":"74996da36ea9d4f382c39e2a617751e9",
            "id":"1123123q",
            "server_time":"1492596274"}
        url = "/getChatMessage?page=1"
        payload = {"id":"123456","data2":"123"}
        headers = {
            'Content-Type':'application/x-www-form-urlencoded',
            'User-Agent':'xxx'
            }
        try:
            rsp = self.client.post(url, data=payload, headers=headers, catch_response=True, cookies=cookies)
            # print r.text
            self.check_response(rsp)
        except Exception as err:
            # logger.exception("request getChatMessage fail:%s", err)
            raise Exception("request getChatMessage fail:{0}".format(err))

class WebsiteUser(HttpLocust):
    task_set = WebsiteTasks
    min_wait = 5000
    max_wait = 15000
    host = "http://aaa.abc.com"