提问人：Nilamber Singh 提问时间：11/7/2023 更新时间：11/13/2023 访问量：31
Locust - 将多个 API 调用组合为一个
Locust - Group multiple api calls into one
问：
有没有办法在蝗虫报告/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")
答：
0赞
Nilamber Singh
11/13/2023
#1
解决了，我所要做的就是从 self.client 中做一个小的更改，以便直接向请求库发出 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")
评论