Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@confluence_measure
def view_dashboard(locust):
params = ViewDashboard()
r = locust.client.get('/index.action', catch_response=True)
content = r.content.decode('utf-8')
keyboard_hash = fetch_by_re(params.keyboard_hash_re, content)
build_number = fetch_by_re(params.build_number_re, content)
if 'quick-search' not in content or 'Log Out' not in content:
logger.error(f'Could not view dashboard: {content}')
assert 'quick-search' in content and 'Log Out' in content, 'Could not view dashboard.'
locust.client.post('/rest/webResources/1.0/resources', params.resources_body.get("205"),
TEXT_HEADERS, catch_response=True)
locust.client.get('/rest/mywork/latest/status/notification/count', catch_response=True)
locust.client.get(f'/rest/shortcuts/latest/shortcuts/{build_number}/{keyboard_hash}', catch_response=True)
locust.client.get(f'/rest/experimental/search?cql=type=space%20and%20space.type=favourite%20order%20by%20'
@confluence_measure
def create_blog():
draft_name = locust.storage['draft_name']
parsed_space_key = locust.storage['parsed_space_key']
content_id = locust.storage['content_id']
atl_token = locust.storage['atl_token']
draft_body = {"status": "current", "title": draft_name, "space": {"key": f"{parsed_space_key}"},
"body": {"editor": {"value": f"Test Performance Blog Page Content {draft_name}",
"representation": "editor", "content": {"id": f"{content_id}"}}},
"id": f"{content_id}", "type": "blogpost",
"version": {"number": 1, "minorEdit": True, "syncRev": "0.mcPCPtDvwoayMR7zvuQSbf8.30"}}
TEXT_HEADERS['Content-Type'] = 'application/json'
r = locust.client.put(f'/rest/api/content/{content_id}?status=draft', json=draft_body,
headers=TEXT_HEADERS, catch_response=True)
content = r.content.decode('utf-8')
if 'current' not in content or 'title' not in content:
@confluence_measure
def search_cql():
r = locust.client.get(f"/rest/api/search?cql=siteSearch~'{generate_random_string(3, only_letters=True)}'"
f"&start=0&limit=20", catch_response=True)
if '{"results":[' not in r.content.decode('utf-8'):
logger.locust_info(r.content.decode('utf-8'))
content = r.content.decode('utf-8')
if 'results' not in content:
logger.error(f"Search cql failed: {content}")
assert 'results' in content, "Search cql failed."
locust.client.get('/rest/mywork/latest/status/notification/count', catch_response=True)
@confluence_measure
def create_page_editor():
r = locust.client.get(f'/pages/createpage.action?spaceKey={space_key}&fromPageId={page_id}&src=quick-create',
catch_response=True)
content = r.content.decode('utf-8')
if 'Page Title' not in content:
logger.error(f'Could not open page editor: {content}')
assert 'Page Title' in content, 'Could not open page editor.'
parsed_space_key = fetch_by_re(params.space_key_re, content)
atl_token = fetch_by_re(params.atl_token_re, content)
content_id = fetch_by_re(params.content_id_re, content)
locust.storage['content_id'] = content_id
locust.storage['atl_token'] = atl_token
locust.client.post('/rest/webResources/1.0/resources', params.resources_body.get("705"),
TEXT_HEADERS, catch_response=True)
@confluence_measure
def open_editor():
create_page_id = locust.storage['create_page_id']
r = locust.client.get(f'/pages/editpage.action?pageId={create_page_id}', catch_response=True)
content = r.content.decode('utf-8')
if '<title>Edit' not in content or 'Update</button>' not in content:
logger.error(f'Could not open PAGE {create_page_id} to edit: {content}')
assert '<title>Edit' in content and 'Update</button>' in content, \
'Could not open PAGE to edit.'
edit_page_version = fetch_by_re(params.editor_page_version_re, content)
edit_atl_token = fetch_by_re(params.atl_token_re, content)
edit_space_key = fetch_by_re(params.space_key_re, content)
edit_content_id = fetch_by_re(params.content_id_re, content)
edit_page_id = fetch_by_re(params.page_id_re, content)
edit_parent_page_id = fetch_by_re(params.parent_page_id, content)</title>
@confluence_measure
def view_attachments(locust):
page = random.choice(confluence_dataset["pages"])
page_id = page[0]
r = locust.client.get(f'/pages/viewpageattachments.action?pageId={page_id}', catch_response=True)
content = r.content.decode('utf-8')
if not('Upload file' in content and 'Attach more files' in content or 'currently no attachments' in content):
logger.error(f'View attachments failed: {content}')
assert 'Upload file' in content and 'Attach more files' in content \
or 'currently no attachments' in content, 'View attachments failed.'
@confluence_measure
def view_page():
r = locust.client.get(f'/pages/viewpage.action?pageId={page_id}', catch_response=True)
content = r.content.decode('utf-8')
if 'Created by' not in content or 'Save for later' not in content:
logger.error(f'Fail to open page {page_id}: {content}')
assert 'Created by' in content and 'Save for later' in content, 'Could not open page.'
parent_page_id = fetch_by_re(params.parent_page_id_re, content)
parsed_page_id = fetch_by_re(params.page_id_re, content)
space_key = fetch_by_re(params.space_key_re, content)
tree_request_id = fetch_by_re(params.tree_result_id_re, content)
has_no_root = fetch_by_re(params.has_no_root_re, content)
root_page_id = fetch_by_re(params.root_page_id_re, content)
atl_token_view_issue = fetch_by_re(params.atl_token_view_issue_re, content)
editable = fetch_by_re(params.editable_re, content)
ancestor_ids = re.findall(params.ancestor_ids_re, content)
@confluence_measure
def edit_page():
locust.storage['draft_name'] = f"{generate_random_string(10, only_letters=True)}"
edit_parent_page_id = locust.storage['edit_parent_page_id']
edit_page_id = locust.storage['edit_page_id']
content_id = locust.storage['edit_content_id']
edit_page_version = int(locust.storage['edit_page_version']) + 1
edit_atl_token = locust.storage['atl_token']
edit_page_body = dict()
if edit_parent_page_id:
edit_page_body = {
"status": "current",
"title": f"Test Performance Edit with locust {locust.storage['draft_name']}",
"space": {
"key": f"{space_key}"
},
@confluence_measure
def like_page(locust):
params = LikePage()
page = random.choice(confluence_dataset["pages"])
page_id = page[0]
JSON_HEADERS['Origin'] = CONFLUENCE_SETTINGS.server_url
r = locust.client.get(f'/rest/likes/1.0/content/{page_id}/likes', headers=JSON_HEADERS, catch_response=True)
content = r.content.decode('utf-8')
like = fetch_by_re(params.like_re, content)
if like is None:
r = locust.client.post(f'/rest/likes/1.0/content/{page_id}/likes', headers=JSON_HEADERS, catch_response=True)
else:
r = locust.client.delete(f'/rest/likes/1.0/content/{page_id}/likes', catch_response=True)
content = r.content.decode('utf-8')
@confluence_measure
def view_page_tree():
tree_request_id = locust.storage['tree_request_id'].replace('&', '&')
ancestors = locust.storage['ancestors']
root_page_id = locust.storage['root_page_id']
viewed_page_id = locust.storage['page_id']
space_key = locust.storage['space_key']
r = ''
# Page has parent
if locust.storage['has_no_root'] == 'false':
request = f"{tree_request_id}&hasRoot=true&pageId={root_page_id}&treeId=0&startDepth=0&mobile=false" \
f"&{ancestors}treePageId={viewed_page_id}&_={timestamp_int()}"
r = locust.client.get(f'{request}', catch_response=True)
# Page does not have parent
elif locust.storage['has_no_root'] == 'true':
request = f"{tree_request_id}&hasRoot=false&spaceKey={space_key}&treeId=0&startDepth=0&mobile=false" \
f"&{ancestors}treePageId={viewed_page_id}&_={timestamp_int()}"