-
Notifications
You must be signed in to change notification settings - Fork 5
/
locust_sequential_tasks.py
151 lines (118 loc) · 4.37 KB
/
locust_sequential_tasks.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
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
from locust import task, HttpUser, SequentialTaskSet
from locust.exception import StopUser
from requests.adapters import HTTPAdapter
from .locustfile import Requests
class SearchTicket(SequentialTaskSet):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.client.mount('https://', HTTPAdapter(pool_maxsize=50))
self.client.mount('http://', HTTPAdapter(pool_maxsize=50))
@task
def only_search(self):
#logging.info("Running task 'only search'...")
task_sequence = ["home_expected", "search_ticket_expected"]
requests = Requests(self.client)
for task in task_sequence:
requests.perform_task(task)
@task
def stop(self):
#logging.info("Stopping task 'only search'...")
raise StopUser()
class BookTicket(SequentialTaskSet):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.client.mount('https://', HTTPAdapter(pool_maxsize=50))
self.client.mount('http://', HTTPAdapter(pool_maxsize=50))
@task
def book_ticket(self):
#logging.info("Running Tasks for booking...")
task_sequence = ["home_expected",
"login_expected",
"search_ticket_expected",
"start_booking_expected",
"get_assurance_types_expected",
"get_foods_expected",
"select_contact_expected",
"finish_booking_expected"]
requests = Requests(self.client)
for task in task_sequence:
requests.perform_task(task)
@task
def stop(self):
#logging.info("Stopping booking tasks")
raise StopUser()
class ConsignTicket(SequentialTaskSet):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.client.mount('https://', HTTPAdapter(pool_maxsize=50))
self.client.mount('http://', HTTPAdapter(pool_maxsize=50))
@task
def perform_task(self):
#logging.debug("Running tasks for 'consign ticket'...")
task_sequence = [
"home_expected",
"login_expected",
"select_contact_expected",
"finish_booking_expected",
"select_order_expected",
"get_consigns_expected",
"confirm_consign_expected",
]
requests = Requests(self.client)
for task in task_sequence:
requests.perform_task(task)
@task
def stop(self):
#logging.info("Stopping consign tasks")
raise StopUser()
class PayForTickets(SequentialTaskSet):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.client.mount('https://', HTTPAdapter(pool_maxsize=50))
self.client.mount('http://', HTTPAdapter(pool_maxsize=50))
@task
def perform_task(self):
# logging.debug("Running tasks for 'pay'...")
task_sequence = ["home_expected",
"login_expected",
"select_contact_expected",
"finish_booking_expected",
"select_order_expected",
"pay_expected"]
requests = Requests(self.client)
for task in task_sequence:
requests.perform_task(task)
@task
def stop(self):
#logging.info("Stopping pay tasks")
raise StopUser()
class CollectTicketTasks(SequentialTaskSet):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.client.mount('https://', HTTPAdapter(pool_maxsize=50))
self.client.mount('http://', HTTPAdapter(pool_maxsize=50))
@task
def perform_task(self):
#logging.debug("Running user 'collect ticket'...")
task_sequence = [
"home_expected",
"login_expected",
"select_order_expected",
"pay_expected",
"collect_ticket_expected",
]
requests = Requests(self.client)
for task in task_sequence:
requests.perform_task(task)
@task
def stop(self):
#logging.info("Stopping collect ticket tasks")
raise StopUser()
class UserGlobal(HttpUser):
tasks = {
SearchTicket: 1,
BookTicket: 1,
ConsignTicket: 1,
PayForTickets: 1,
CollectTicketTasks: 1
}