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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
from locust import FastHttpUser, run_single_user, task
from locust.contrib.fasthttp import RestResponseContextManager
from locust.user.wait_time import constant
from contextlib import contextmanager
from typing import Generator
class MyUser(FastHttpUser):
host = "https://postman-echo.com"
wait_time = constant(180) # be nice to postman-echo.com, and dont run this at scale.
@task
def t(self):
# should work
with self.rest("GET", "/get", json={"foo": 1}) as resp:
if resp.js["args"]["foo"] != 1:
resp.failure(f"Unexpected value of foo in response {resp.text}")
# should work
with self.rest("POST", "/post", json={"foo": 1}) as resp:
if resp.js["data"]["foo"] != 1:
resp.failure(f"Unexpected value of foo in response {resp.text}")
# assertions are a nice short way to express your expectations about the response. The AssertionError thrown will be caught
# and fail the request, including the message and the payload in the failure content.
assert resp.js["data"]["foo"] == 1, "Unexpected value of foo in response"
# assertions are a nice short way to validate the response. The AssertionError they raise
# will be caught by rest() and mark the request as failed
with self.rest("POST", "/post", json={"foo": 1}) as resp:
# mark the request as failed with the message "Assertion failed"
assert resp.js["data"]["foo"] == 2
with self.rest("POST", "/post", json={"foo": 1}) as resp:
# custom failure message
assert resp.js["data"]["foo"] == 2, "my custom error message"
with self.rest("POST", "/post", json={"foo": 1}) as resp:
# use a trailing comma to append the response text to the custom message
assert resp.js["data"]["foo"] == 2, "my custom error message with response text,"
# this only works in python 3.8 and up, so it is commented out:
# if sys.version_info >= (3, 8):
# with self.rest("", "/post", json={"foo": 1}) as resp:
# # assign and assert in one line
# assert (foo := resp.js["foo"])
# print(f"the number {foo} is awesome")
# rest() catches most exceptions, so any programming mistakes you make automatically marks the request as a failure
# and stores the callstack in the failure message
with self.rest("POST", "/post", json={"foo": 1}) as resp:
1 / 0 # pylint: disable=pointless-statement
# response isnt even json, but RestUser will already have been marked it as a failure, so we dont have to do it again
with self.rest("GET", "/") as resp:
pass
with self.rest("GET", "/") as resp:
# If resp.js is None (which it will be when there is a connection failure, a non-json responses etc),
# reading from resp.js will raise a TypeError (instead of an AssertionError), so lets avoid that:
if resp.js:
assert resp.js["foo"] == 2
# or, as a mildly confusing oneliner:
assert not resp.js or resp.js["foo"] == 2
# 404
with self.rest("GET", "http://example.com/") as resp:
pass
# connection closed
with self.rest("POST", "http://example.com:42/", json={"foo": 1}) as resp:
pass
# An example of how you might write a common base class for an API that always requires
# certain headers, or where you always want to check the response in a certain way
class RestUserThatLooksAtErrors(FastHttpUser):
abstract = True
@contextmanager
def rest(self, method, url, **kwargs) -> Generator[RestResponseContextManager, None, None]:
extra_headers = {"my_header": "my_value"}
with super().rest(method, url, headers=extra_headers, **kwargs) as resp:
if resp.js and "error" in resp.js and resp.js["error"] is not None:
resp.failure(resp.js["error"])
yield resp
class MyOtherRestUser(RestUserThatLooksAtErrors):
host = "https://postman-echo.com"
wait_time = constant(180) # be nice to postman-echo.com, and dont run this at scale.
@task
def t(self):
with self.rest("GET", "/") as _resp:
pass
if __name__ == "__main__":
run_single_user(MyUser)
|
