Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from locust import HttpLocust, task, TaskSet
class ModelTaskSet(TaskSet):
@task()
def invoke_nine_decode(self):
self._model_client.invoke(uid=1)
def on_start(self):
self._model_client = model.ModelClient(model.calculate_url_from_config(),
token=edi.build_client().get_token(config.MODEL_DEPLOYMENT_NAME),
http_client=self.client)
class TestLocust(HttpLocust):
task_set = ModelTaskSet
min_wait = 0
max_wait = 0
response_length=0)
break
elif (game_server_state == "UnAllocated"):
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(
request_type="GameServerUnAllocated",
name="GameServerUnAllocated",
response_time=total_time,
response_length=0)
break
else:
self_link = response_json['metadata']['selfLink']
response = self.client.get(self_link)
class AgonesUser(HttpLocust):
def setup(self):
# Create a fleet.
client = clients.HttpSession(base_url=self.host)
self.create_fleet(client, FLEET_NAME, FLEET_SIZE)
def create_fleet(self, client, fleet_name, fleet_size):
# Create a Fleet and wait for it to scale up.
print "Creating Fleet: " + fleet_name
payload = {
"apiVersion": "agones.dev/v1",
"kind": "Fleet",
"metadata": {
"name": str(fleet_name),
"namespace": "default"
},
"spec": {
def on_start(self):
self.client.post("/login", {
"username": "test_user",
"password": ""
})
@task
def index(self):
self.client.get("/")
@task
def about(self):
self.client.get("/about/")
class WebsiteUser(HttpLocust):
task_set = WebsiteTasks
min_wait = 100
max_wait = 1500
class WebsiteTasks(TaskSet):
def on_start(self):
print("start...")
@task(1)
def index(self):
self.client.get("/hello-world")
# @task(1)
# def about(self):
# self.client.get("/hello")
class WebsiteUser(HttpLocust):
task_set = WebsiteTasks
host = "http://127.0.0.1:9988"
min_wait = 1000
max_wait = 5000
if status == 'revoked':
self._user_credentials.remove(credential_uuid)
self.user_credential_client.credentials(credential_uuid).patch(data=data, name='/api/v2/credentials/[uuid]')
@task(10)
def render_credential(self):
""" Render a credential. """
if not self._user_credentials:
return
credential_uuid = random.choice(self._user_credentials)
url = '{}/credentials/{}/'.format(CREDENTIAL_SERVICE_URL.strip('/'), credential_uuid)
self.client.get(url, name='/credentials/[uuid]')
class CredentialsLocust(HttpLocust):
"""Representation of an HTTP "user" to be hatched.
Hatched users will be used to attack the system being load tested. This class
defines which TaskSet class should define the user's behavior and how long a simulated
user should wait between executing tasks. This class also provides a custom client used
to interface with edX REST APIs.
"""
min_wait = 30
max_wait = 100
task_set = CredentialTaskSet
from locust import TaskSet, task, HttpLocust
from locust.clients import HttpSession
from locust.exception import LocustError
from helpers.api import LocustEdxRestApiClient
from programs.config import PROGRAMS_API_URL, JWT_AUDIENCE, JWT_ISSUER, JWT_EXPIRATION_DELTA, JWT_SECRET_KEY
class ProgramsTaskSet(TaskSet):
"""Tasks exercising Programs functionality."""
@task
def list_programs(self):
self.client.programs.get()
class ProgramsUser(HttpLocust):
"""Representation of an HTTP "user" to be hatched.
Hatched users will be used to attack the system being load tested. This class
defines which TaskSet class should define the user's behavior and how long a simulated
user should wait between executing tasks. This class also provides a custom client used
to interface with edX REST APIs.
"""
# Django's AbstractUser, subclassed by Programs, limits usernames to 30 characters.
# See: https://github.com/django/django/blob/stable/1.8.x/django/contrib/auth/models.py#L385
USERNAME_MAX_LENGTH = 30
USERNAME_PREFIX = 'load-test-'
task_set = ProgramsTaskSet
min_wait = 30
max_wait = 100
@task(50)
def load_page(self, url=None):
url = random.choice(self.toc_urls)
r = self.client.get(url)
pq = PyQuery(r.content)
link_elements = pq("a.internal")
self.urls_on_current_page = [l.attrib["href"] for l in link_elements]
@task(30)
def load_sub_page(self):
url = random.choice(self.urls_on_current_page)
r = self.client.get(url)
class AwesomeUser(HttpLocust):
task_set = BrowseDocumentation
host = "http://docs.locust.io/en/latest/"
# we assume someone who is browsing the Locust docs,
# generally has a quite long waiting time (between
# 20 and 600 seconds), since there's a bunch of text
# on each page
min_wait = 2 * 1000
max_wait = 6 * 1000
@task
def test_any(self):
test_dict = random.choice(self.locust.tests)
try:
self.test_runner.run_test(test_dict)
except (AssertionError, MyBaseError, MyBaseFailure) as ex:
request_failure.fire(
request_type=self.test_runner.exception_request_type,
name=self.test_runner.exception_name,
response_time=0,
exception=ex,
)
class WebPageUser(HttpLocust):
host = ""
task_set = WebPageTasks
min_wait = 10
max_wait = 30
# file_path is generated on locusts startup
file_path = "$TESTCASE_FILE"
tests = prepare_locust_tests(file_path)
wait_time = lambda self: random.expovariate(1)
task_set = UserTasks
def strictExp(min_wait,max_wait,mu=1):
"""
Returns an exponentially distributed time strictly between two bounds.
"""
while True:
x = random.expovariate(mu)
increment = (max_wait-min_wait)/(mu*6.0)
result = min_wait + (x*increment)
if result