Locust - 将多个 api 调用分组为一个

Question

有没有办法在 Locust 报告/UI 中将多个 api 调用分组为 1？我有这个特定的要求，因为我想实现 40 个事务/秒，在我的例子中，3 个 api 调用以顺序方式组成 1 个事务。我已经将所有 api 调用包装到 1 个任务中。这是代码。

    host = config_utils.get_from_test_config('platform.initiate.baseUri')

    def on_start(self):
        self.account_host = config_utils.get_from_test_config('platform.initiate.baseUri')
        self.api_host = config_utils.get_from_test_config('platform.api.baseUri')

        self.username = config_utils.get_from_merchant_config('customer.go.apiToken')
        self.password = config_utils.get_from_merchant_config('customer.go.apiSecret')

    @task
    def complete_workflow(self):
        self.client.locust_name = "complete_workflow"
        access_token = oauth.get_oauth_token(self.username, self.password)
        initiate_headers = {'Content-Type': 'application/json', "Authorization": f"Bearer {access_token}"}
        payload = {"customerInternalReference": "QEC PENN testing customer", "workflowDefinition": {"key": 3}}
        initiate_response = self.client.post(f"{self.account_host}/api/v1/accounts", json=payload,
                                             headers=initiate_headers, name="v3")

        response = json.loads(initiate_response.text)

        workflow_credentials = response.get("workflowExecution", {}).get("credentials", [])

        id_credentials = [credential for credential in workflow_credentials if credential.get("category") == "ID"]

        selfie_credentials = [credential for credential in workflow_credentials if
                              credential.get("category") == "SELFIE"]

        self.workflow_id = response.get("workflowExecution", {}).get("id")
        self.account_id = response.get("account", {}).get("id")
        self.api_token = id_credentials[0].get("api", {}).get("token")
        self.id_credential = id_credentials[0].get("id")
        self.selfie_credential = selfie_credentials[0].get("id")

        front_image = (
            'USA_DL_front.jpg',
            open('images/USA_DL_front.jpg', 'rb'),
            "image/jpeg"
        )

        data = {'file': front_image}

        encoder = MultipartEncoder(fields=data)
        headers = {'Accept': '*/*', 'Content-Type': encoder.content_type, "Authorization": f"Bearer {self.api_token}"}

        self.client.post(
            f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}/credentials/{self.id_credential}/parts/FRONT",
            data=encoder, headers=headers, name="v3")


        finalise_header = {'Accept': '*/*', "Authorization": f"Bearer {self.api_token}"}
        self.client.put(
            f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}",
            headers=finalise_header, name="finalise")

Answer 1

解决了，我所要做的就是从 self.client 进行一个小更改，以便直接向 requests 库发出 api 请求，以便 Locust 无法跟踪这些调用，因此不会记录在报告中。我只是使用 self.client 进行最后一次 API 调用，这样只有最后一次调用才会记录在报告中，假设最后一次调用成功，交易也成功。

host = config_utils.get_from_test_config('platform.initiate.baseUri')

def on_start(self):
    self.account_host = config_utils.get_from_test_config('platform.initiate.baseUri')
    self.api_host = config_utils.get_from_test_config('platform.api.baseUri')

    self.username = config_utils.get_from_merchant_config('customer.go.apiToken')
    self.password = config_utils.get_from_merchant_config('customer.go.apiSecret')

@task
def complete_workflow(self):
    self.client.locust_name = "complete_workflow"
    access_token = oauth.get_oauth_token(self.username, self.password)
    initiate_headers = {'Content-Type': 'application/json', "Authorization": f"Bearer {access_token}"}
    payload = {"customerInternalReference": "QEC PENN testing customer", "workflowDefinition": {"key": 3}}
    initiate_response = requests.post(f"{self.account_host}/api/v1/accounts", json=payload,
                                         headers=initiate_headers)

    response = json.loads(initiate_response.text)

    workflow_credentials = response.get("workflowExecution", {}).get("credentials", [])

    id_credentials = [credential for credential in workflow_credentials if credential.get("category") == "ID"]

    selfie_credentials = [credential for credential in workflow_credentials if
                          credential.get("category") == "SELFIE"]

    self.workflow_id = response.get("workflowExecution", {}).get("id")
    self.account_id = response.get("account", {}).get("id")
    self.api_token = id_credentials[0].get("api", {}).get("token")
    self.id_credential = id_credentials[0].get("id")
    self.selfie_credential = selfie_credentials[0].get("id")

    front_image = (
        'USA_DL_front.jpg',
        open('images/USA_DL_front.jpg', 'rb'),
        "image/jpeg"
    )

    data = {'file': front_image}

    encoder = MultipartEncoder(fields=data)
    headers = {'Accept': '*/*', 'Content-Type': encoder.content_type, "Authorization": f"Bearer {self.api_token}"}

    requests.post(
        f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}/credentials/{self.id_credential}/parts/FRONT",
        data=encoder, headers=headers)


    finalise_header = {'Accept': '*/*', "Authorization": f"Bearer {self.api_token}"}
    self.client.put(
        f"{self.api_host}/api/v1/accounts/{self.account_id}/workflow-executions/{self.workflow_id}",
        headers=finalise_header, name="finalise")