Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use total_rps method in locust

Best Python code snippet using locust

test.py

Source:test.py

1#!/usr/bin/env python2import argparse3import subprocess4import sys5import json6from operator import itemgetter7from collections import OrderedDict8import random9FRAMEWORKS = ['lumen', 'phalcon', 'silex', 'slim']10random.shuffle(FRAMEWORKS)11KEYS = {12	'Time taken for tests': 'time_taken',13	'Complete requests': 'complete',14	'Failed requests': 'failed',15	'Requests per second': 'rps',16	'Time per request': 'tpr'17}18SCORES = ['1st', '2nd', '3rd', '4th']19def getdata(req):20	data = {}21	for line in req.splitlines():22		line = [x.strip() for x in line.split(':', 2)]23		if line[0] in KEYS.keys():24			key = KEYS[line[0]]25			if key == 'tpr':26				s = float(line[1].split('[ms]', 2)[0].strip())27				28				if 'tpr' in data:29					data['tpr_concurrent'] = s30				else:31					data['tpr'] = s32			elif key == 'rps':33				data[key] = float(line[1].split('[#/sec]', 2)[0].strip())34			elif key == 'time_taken':35				data[key] = float(line[1].split(' ', 2)[0].strip())36			elif key == 'complete' or key == 'failed':37				data[key] = int(line[1])38	return data39parser = argparse.ArgumentParser(description='PAFB install')40choices = ['all']41choices.extend(FRAMEWORKS)42parser.add_argument("-f", "--frameworks", 43                    help="Which frameworks to test", 44                    choices=choices, 45                    nargs='+',46                    default='all')47parser.add_argument("-n", "--number", 48                    help="Number of requests to make", 49                    type=int,50                    default=1000)51parser.add_argument("-c", "--concurrent", 52                    help="Number of multiple requests to make at a time", 53                    type=int,54                    default=100)55args = parser.parse_args()56if args.frameworks == 'all' or 'all' in args.frameworks:57	args.frameworks = FRAMEWORKS58try:59    subprocess.check_output("which ab", shell=True)60except subprocess.CalledProcessError:61	sys.exit('Install ApacheBench (https://httpd.apache.org/docs/2.4/programs/ab.html) then run this tester again')62print 'Welcome to the PHP API Framework benchmark (PAFB)'63print ''64tests = OrderedDict()65tests['insert'] = 'ab -p tests/insert -n {} -c {} http://{}.pafb.dev:80/'66tests['update'] = 'ab -p tests/update -n {} -c {} http://{}.pafb.dev:80/aaa'67tests['select'] = 'ab -n {} -c {} http://{}.pafb.dev:80/bbb'68tests['delete'] = 'ab -p tests/insert -n {} -c {} http://{}.pafb.dev:80/delete'69tests['index'] = 'ab  -n {} -c {} http://{}.pafb.dev:80/'70results = {}71winners = {framework: {'score': 0, 'total_rps': 0, 'name': framework} for framework in FRAMEWORKS}72for test_key, test_value in tests.iteritems():73	results[test_key] = []74	print 'Testing: {}'.format(test_key.title())75	for framework in FRAMEWORKS:76		req = test_value.format(args.number, args.concurrent, framework)77		test = subprocess.check_output(req, shell=True, stderr=subprocess.STDOUT)78		79		data = getdata(test)80		data.update({'name': framework})81		if data['failed'] > 0:82			print 'Test failed for {}'.format(framework)83			data['rps'] = 084		results[test_key].append(data)85	results[test_key] = sorted(results[test_key], key=itemgetter('rps'), reverse=True)86	print "\nResults:"87	for index, result in enumerate(results[test_key]):88		print '{}. {} - {} requests per second'.format(SCORES[index], result['name'].title(), result['rps'])89	print ''90framework_len = len(FRAMEWORKS)91for result in results.values():92	for index, framework in enumerate(result):93		winners[framework['name']]['score'] += framework_len - index94		winners[framework['name']]['total_rps'] = round(winners[framework['name']]['total_rps'] + framework['rps'], 2)95winners = sorted(winners.values(), key=itemgetter('score', 'total_rps'), reverse=True)96print "Overall Results"97for index, winner in enumerate(winners):...

calc_thrput.py

Source:calc_thrput.py

1import math2import os3import sys4with open(str(sys.argv[1]), "r") as f:5    total_rps = 0.06    total_thp = 0.07    count = 08    for line in f:9        res = line.split()10        if len(res) == 4:11           total_rps += float(res[1])12           total_thp += float(res[3])13           count += 114    f.close()15    avg_rps = round(total_rps/float(count), 4)16    avg_thp = round(total_thp/float(count), 4)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.