1 from locust import events
2 from gevent._semaphore import Semaphore
3 all_locusts_spawned = Semaphore()
4 all_locusts_spawned.acquire()
5
6 def on_hatch_complete(**kwargs):
7 all_locusts_spawned.release() ##创建钩子方法
8
9
10 # 挂在到locust钩子函数（所有的Locust实例产生完成时触发）
11 # events.hatch_complete += on_hatch_complete # 1.0之前的写法
12 events.spawning_complete.add_listener(on_hatch_complete) # 1.0之后的写法
1 # -*- coding: utf-8 -*-
2 # Author : Ethan
3 # Time : 2023/3/30 21:20
4 from locust import HttpUser,task,between,constant,constant_pacing,tag,User
5 import urllib3
6 import requests,uuid,time
7 import requests
8 urllib3.disable_warnings()
9 import random,os,queue
10 from Locust_gather_**** import *
11
12 class user_por(HttpUser):
13 wait_time =between(0,0)
14 @task
15 def get_user_labels(self):
16 self.user_id = str(int(time.time() * 10000000)) + str(random.randint(1, 1000))
17 self.url='/feign/user/labels'
18 self.data= [{"userId":self.user_id,"firstDimension":"TAKEOUT","thirdDimension":"","secondDimension":"STOREID:18353"}]
19 self.header={
20 'Content-Type': 'application/json'
21 }
22 all_locusts_spawned.wait()
23 with self.client.post(url=self.url,headers=self.header,json=self.data,verify=False ,catch_response=True) as response:
24 try:
25 userLabels =response.json()[0]['userLabels']
26 except Exception as e:
27 response.failure(response.content)
28 print(self.user_id)
29 else:
30 if "TAKEOUT_STORE_" in str(userLabels):
31 response.success()
32 else:
33 response.failure(response.json())
34
35 def on_start(self):
36 pass
37 def on_stop(self):
38 pass
39
40 if __name__ == '__main__':
41 os.system('locust -f user_portrait.py --host=http://*****.cf93fe1dde8584346a2d81575c79aff9e.cn-shenzhen.alicontainer.com')