-
Notifications
You must be signed in to change notification settings - Fork 0
/
template.j2
123 lines (112 loc) · 3.67 KB
/
template.j2
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
# Generated using Jinja2
# site {{ site_name }}
# site {{ site_uuid }}
# org {{ org_name }}
# org {{ org_uuid }}
import os
import string
import random
import time
import pytest
import json
import selenium
from locust import HttpUser, TaskSet, task, events
from locust.clients import HttpSession
from pyquery import PyQuery
{%- if pytest_flag %}
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.common.keys import Keys
{%- endif %}
{% if pytest_flag %}
# include PyTest Scripts here
class {{ pytest_class }}():
def setup_method(self, method):
options = webdriver.ChromeOptions()
options.add_argument('headless')
self.driver = webdriver.Chrome(chrome_options=options)
self.vars = {}
def teardown_method(self, method):
self.driver.quit()
{% for code in pytest_codes %}
def {{ code.name }}(self):
{%- for line in code.lines %}
{{ line|replace('"', '"') }}
{%- endfor %}
{%- endfor %}
{% endif %}
class MyTaskSet(TaskSet):
# Include PyTest Files
{%- if pytest_flag %}
{%- for test in pytest_files %}
@task({{ test.ratio }})
def pytest_{{ test.name }}(self):
start_time = time.time()
os.system("pytest pytest/{{ test.path }}")
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(
request_type="pytest",
name="{{ test.name }}",
response_time=total_time,
response_length=0
)
{%- endfor %}
{%- endif %}
# Custom Locust Steps
{%- for sequence in custom_sequences %}
@task({{ sequence.ratio }})
def custom_{{ sequence.name }}(self):
start_time = time.time()
{%- for step in sequence.steps %}
{%- if step.type == "get" %}
self.client.get("{{step.path}}")
{%- elif step.type == "post" %}
self.client.post("{{ step.path}}", {{ step.data|replace('"', '"') }})
{%- endif %}
{%- endfor %}
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(
request_type="locust_steps",
name="{{ sequence.request_name }}",
response_time=total_time,
response_length=0
)
{%- endfor %}
# Top Visited Files
{% for path in top_visited %}
@task({{ path.ratio }})
def path_homepage(self):
self.client.get("{{ path.url_path }}")
{% endfor %}
# Anonymous pages
{% if random %}
def anon_index_page(self, path="/"):
r = self.client.get("/")
pq = PyQuery(r.content)
link_elements = pq("a")
self.urls_on_current_page = []
for l in link_elements:
if "href" in l.attrib:
url = l.attrib["href"]
if url != "/user/logout":
#ensure internal link
if url != "//{{ domain }}" and url is not None and "#" not in url and (url.startswith('{{ url }}') or url.startswith('/')):
self.urls_on_current_page.append(l.attrib["href"])\
@task({{ random_ratio }})
def load_anon_index_page(self):
self.anon_index_page("/")
count = 0
while (count < 10):
count = count + 1
url = random.choice(self.urls_on_current_page)
print("Calling ", url)
self.client.get(url)
{% endif %}
class MyLocust(HttpUser):
host = os.getenv('TARGET_URL', "{{ url }}")
task_set = MyTaskSet
min_wait = 1000
max_wait = 3000