Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@seq_task(1)
def click_login(self):
global user_number
user_number += 1
self.user_name = 'JohnDoe' + str(user_number)
self.user_email = self.user_name + '@john.doe'
self.user_home_page = '/users/' + self.user_name
print('==== starting user = %s ====' % self.user_name)
response = self.client.get("/+login")
if response.status_code != 200:
print('%s: response.status_code = %s' % (sys._getframe().f_lineno, response.status_code))
@seq_task(5)
def click_post_login(self):
response = self.client.post("/+login",
{"login_username": self.user_name,
"login_password": "locust123",
"login_submit": "1",
"login_nexturl": "/Home",
})
if response.status_code != 200:
print('%s: response.status_code = %s' % (sys._getframe().f_lineno, response.status_code))
@seq_task(8)
def create_new_page(self):
# first get yields 404 status code
new_item_name = 'Locust-' + self.user_name
response = self.client.get('/' + new_item_name, catch_response=True)
if response.status_code == 404:
response.success()
else:
print('%s: response.status_code = %s' % (sys._getframe().f_lineno, response.status_code))
# click MoinMoin markup link
page_get = '/+modify/' + new_item_name + '?contenttype=text%2Fx.moin.wiki%3Bcharset%3Dutf-8&itemtype=default'
response = self.client.get(page_get)
if response.status_code != 200:
print('%s: response.status_code = %s' % (sys._getframe().f_lineno, response.status_code))
# complete form and post
new_content = '= %s =\n\nNew Item created by Locust' % new_item_name
new_page_post = '/+modify/' + new_item_name + '?contenttype=text%2Fx.moin.wiki%3Bcharset%3Dutf-8&itemtype=default&template='
@seq_task(4)
def cancel_session_token(self):
self.client.call_endpoint(
'CancelSessionToken',
dict(session_token=self.session_token),
)
@seq_task(2)
def click_register(self):
response = self.client.get("/+register")
if response.status_code != 200:
print('%s: response.status_code = %s' % (sys._getframe().f_lineno, response.status_code))
@seq_task(9)
@task(10)
def modify_new_page(self):
# click link to new page
new_item_name = 'Locust-' + self.user_name
response = self.client.get('/' + new_item_name)
if response.status_code != 200:
print('%s: response.status_code = %s' % (sys._getframe().f_lineno, response.status_code))
# click MoinMoin markup link
page_get = '/+modify/' + new_item_name + '?contenttype=text%2Fx.moin.wiki%3Bcharset%3Dutf-8&itemtype=default'
response = self.client.get(page_get)
if response.status_code != 200:
print('%s: response.status_code = %s' % (sys._getframe().f_lineno, response.status_code))
# complete form and post
new_content = '= %s =\n\nNew Item created by Locust' % new_item_name
new_page_post = '/+modify/' + new_item_name + '?contenttype=text%2Fx.moin.wiki%3Bcharset%3Dutf-8&itemtype=default&template='
response = self.client.post(new_page_post,
@seq_task(7)
def save_messages(self):
"""save messages left over when last update_home_page fails due to lockout"""
if self.message:
response = self.client.get('/+modify/' + self.user_name + '?contenttype=text%2Fx.moin.wiki%3Bcharset%3Dutf-8&itemtype=default')
if response.status_code != 200:
print('%s: response.status_code = %s' % (sys._getframe().f_lineno, response.status_code))
# complete form and post
new_content = '= %s =\n\n== Unsaved locked out messages ==%s' % (self.user_name, self.message)
home_page_post = '/+modify/' + self.user_name + '?contenttype=text%2Fx.moin.wiki%3Bcharset%3Dutf-8&itemtype=default&template='
response = self.client.post(home_page_post,
{"content_form_data_text": new_content,
"comment": "my comment",
"submit": "OK",
'meta_form_contenttype': 'text/x.moin.wiki;charset=utf-8',
"meta_form_itemtype": "default",
"meta_form_acl": "None",
@seq_task(2)
@task(1)
def getProducts(self):
logging.info("Auth User - Get Catalog")
self.client.get("/products")
@seq_task(4)
@task(2)
def addToCart(self):
self.listCatalogItems()
productid = random.choice(products)
logging.info("Add to Cart for user " + self.locust.userid)
cart = self.client.post("/cart/item/add/" + self.locust.userid, json={
"name": productid,
"price": "100",
"shortDescription": "Test add to cart",
"quantity": random.randint(1,2),
"itemid": productid
})
products.clear()
@seq_task(3)
@task(2)
def getProduct(self):
logging.info("Auth User - Get a product")
products = self.listCatalogItems()
id = random.choice(products)
product = self.client.get("/products/"+ id).json()
logging.info("Product info - " + str(product))
products.clear()