performance-testing - 我们如何根据请求数终止 locust 进程？

Question

我必须使用 locust.io 对多个 API 进行负载测试，所以我想为每个 API 执行 10 个请求（假设我有 10 个 API，所以每个 API 应该得到 10 个请求）并且应该自动终止进程如果所有请求都已完成。

下面是我尝试使用 for 循环但尚未工作的代码片段。

class UserBehavior(TaskSet):

    def __init__(self, parent):
        super().__init__(parent)
        self.camp_code = random.choice(t.camp_code)
        self.pgm_id = random.choice(t.pgm_id)


    @task
    def alloction_details(self):

        pgm_id = self.pgm_id
        camp_code = self.camp_code
        print("L21",camp_code)

        myheaders = {'Content-Type': 'application/json', 'Accept': 'application/json'}

        load = {
            "Code": str(camp_code),
            "Id": str(pgm_id)}
        # print("L43", users)

        for i in range(10):
            
            if i==9:
                self.interrupt()              
            else:
                with self.client.post('/ContentDetails', data= json.dumps(load), headers=myheaders,catch_response=True) as response:
                    print(response.status_code)
                    if response.status_code==200:
                        response.success()
                        # res = response.json()
                        print("Response of allocation details ", response.text, '\n\n')
                    else:
                        response.failure('API failure')

            
    @task
    def get_content_details(self):

        myheaders = {'Content-Type': 'application/json', 'Accept': 'application/json'}
        pgm_id = self.pgm_id
        camp_code = self.camp_code

        for i in range(10):

            if i == 9: 
                # sleep(1000)
                self.interrupt()
            else:
                with self.client.post('/ScheduleDetails?Code='+str(camp_code), catch_response=True) as response:
                    print(response.status_code)

                    if response.status_code==200:
                        response.success()
                        # res = response.json()
                        print("Response : ", response.text, '\n\n')
                    else:
                        response.failure('API failure')  

class WebsiteUser(HttpUser):

    tasks = [UserBehavior]
    host = 'localhost'
    wait_time = between(1, 3)

标签： performance-testinglocust