-
Notifications
You must be signed in to change notification settings - Fork 1
/
locust.py
63 lines (50 loc) · 1.47 KB
/
locust.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import contextlib
import io
import os
import random
import time
import uuid
import urllib
import requests
from locust import HttpLocust, TaskSet, task, between
import json
TARGET_IMAGE_FILES = [
# XXX: from https://randomuser.me/photos
*[f"https://randomuser.me/api/portraits/men/{idx}.jpg" for idx in range(100)],
*[f"https://randomuser.me/api/portraits/women/{idx}.jpg" for idx in range(100)],
# XXX: from https://tinyfac.es/
]
FETCHED_IMAGE_FILES = {}
class UserTaskSet(TaskSet):
@task
def send(self):
req_id = str(uuid.uuid4()) # XXX: Just for tracing
picked_image = random.choice(TARGET_IMAGE_FILES)
taskname = 'predict/url'
headers = {'content-type': 'application/json'}
payload = {"url": picked_image }
started = time.time()
response = self.client.post(f"{taskname}", data=json.dumps(payload), headers=headers)
ended = time.time()
duration = ended - started
print(
(
"out",
req_id,
taskname,
duration,
response.status_code,
len(response.content),
)
)
TARGET_RPS = 10
class User(HttpLocust):
task_set = UserTaskSet
def wait_time(self):
target_wait = between(0, 1 / TARGET_RPS)(self)
print(("wait", target_wait))
return target_wait
if __name__ == "__main__":
ts = UserTaskSet()
ts.client = requests
ts.send()