KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use get_rest_data method in yandex-tank
Best Python code snippet using yandex-tank
Nova.py
Source:Nova.py 
...8import time9class Nova(OpenstackService):10 def __init__(self):11 OpenstackService.__init__(self)12 def get_rest_data(self, url):13 try:14 return self.restful.get_req(url)15 except:16 print "Token expires, update it now..."17 self.update_token()18 return self.restful.get_req(url)19 def post_rest_data(self, url, data):20 try:21 return self.restful.post_req(url, data)22 except:23 print "Token expires, update it now..."24 self.update_token()25 return self.restful.post_req(url, data)26 def getInstances(self):27 """28 get instances that belong to a tenant29 """30 url = "%s/v2/%s/servers" % (OpenstackConf.NOVA_URL, self.tenantId)31 result = self.get_rest_data(url)32 servers = result['servers']33 instances = []34 for s in servers:35 instances.append(str(s['id']))36 return instances37 def get_id_from_name(self, name):38 """39 get instances id with specific name40 """41 url = "%s/v2/%s/servers" % (OpenstackConf.NOVA_URL, self.tenantId)42 servers = self.get_rest_data(url)['servers']43 for s in servers:44 if(str(s['name']) == name):45 return str(s['id'])46 return None47 def get_host_from_vid(self, vid):48 url = "{0}/v2/{1}/servers/{2}".format(OpenstackConf.NOVA_URL, self.tenantId, vid)49 info = self.get_rest_data(url)['server']50 return info["OS-EXT-SRV-ATTR:host"]51 def getInstancesOnHost(self, host):52 url = "%s/v2/%s/servers?host=%s" % (OpenstackConf.NOVA_URL, self.tenantId, host)53 result = self.get_rest_data(url)54 servers = result['servers']55 instances = []56 for s in servers:57 instances.append(str(s['id']))58 return instances59 def getComputeHosts(self):60 url = "%s/v2/%s/os-hosts" % (OpenstackConf.NOVA_URL, self.tenantId)61 result = self.get_rest_data(url)62 hostsList = result['hosts']63 hosts = []64 for host in hostsList:65 if host['service'] == 'compute':66 hosts.append(str(host['host_name']))67 return hosts68 def inspect_host(self, host):69 """70 Get the detail msg of a specific host71 :param host: host to inspect72 :return: dict info of a host73 """74 url = "%s/v2/%s/os-hosts/%s" % (OpenstackConf.NOVA_URL, self.tenantId, host)75 results = self.get_rest_data(url)['host']76 assert results[0]['resource']['project'] == '(total)'77 assert results[1]['resource']['project'] == '(used_now)'78 info = dict()79 info['cpu'] = {'total': results[0]['resource']['cpu'], 'used': results[1]['resource']['cpu']}80 info['mem'] = {'total': results[0]['resource']['memory_mb'], 'used': results[1]['resource']['memory_mb']}81 info['disk'] = {'total': results[0]['resource']['disk_gb'], 'used': results[1]['resource']['disk_gb']}82 return info83 def inspect_flavor(self, flavor_id):84 """85 Get openstack flavor info by its id, flavor contains vm's config of cpu virtual core num, ram and disk size86 :param flavor_id: id of the flavor87 :return: flavor info, disk in GB, RAM in MB, vpus num88 """89 url = "{0}/v2/{1}/flavors/{2}".format(OpenstackConf.NOVA_URL, self.tenantId, flavor_id)90 results = self.get_rest_data(url)['flavor']91 return results92 def inspect_instance(self, vm_uuid):93 """94 Get detail info of a instance95 :param vm_uuid: the uuid of the instance96 :return: dict info of the instance97 """98 url = "{0}/v2/{1}/servers/{2}".format(OpenstackConf.NOVA_URL, self.tenantId, vm_uuid)99 info = self.get_rest_data(url)['server']100 flavor_id = info['flavor']['id']101 flavor_info = self.inspect_flavor(flavor_id)102 info['disk'] = flavor_info['disk']103 info['cpu'] = flavor_info['vcpus']104 info['mem'] = flavor_info['ram']105 return info106 107 def resize_instance(self, instance_id, flavor_id):108 """ resize a given instance to another flavor with id=flavor_id109 @param instance_id:110 @param host:111 """112 url = "{base}/v2/{tenant}/servers/{instance}/action".format(base=OpenstackConf.NOVA_URL,113 tenant=self.tenantId, instance=instance_id)...
rest_user_service.py
Source:rest_user_service.py
...16user_service = UserService(mykafka.create_producer('ec2-35-159-21-220.eu-central-1.compute.amazonaws.com', 9092))17@app.route('/add_user', methods=['POST'])18def restAddUser():19 try:20 data = rest_common.get_rest_data(request)21 user_service.add_user(data)22 return rest_common.create_response(200)23 except InvalidActionException as e:24 return rest_common.create_response(400, e.toDict())25 except UserExistsException as e:26 return rest_common.create_error_response(409, e)27 except CommandFailedException as e:28 return rest_common.create_error_response(504, e)29@app.route('/authenticate_user', methods=['POST'])30def restAuthenticateUser():31 try:32 data = rest_common.get_rest_data(request)33 session_id = user_service.authenticate_user(data)34 return rest_common.create_response(200, {'session_id':str(session_id)})35 except InvalidActionException as e:36 return rest_common.create_response(400, e.toDict())37 except UserUnknownException as e:38 return rest_common.create_error_response(404, e)39 except InvalidPasswortException as e:40 return rest_common.create_error_response(401, e)41 except CommandFailedException as e:42 return rest_common.create_error_response(504, e)43#@app.route('/update_user_adress', methods=['POST'])44#def restUpdateAdress():45# try:46# data = rest_common.get_rest_data(request)47# user_service.update_user_adress(data)48# return rest_common.create_response(200)49# except InvalidActionException as e:50# return rest_common.create_error_response(400, e)51# except UserUnknownException as e:52# return rest_common.create_error_response(404, e)53# except SessionElapsedException as e:54# return rest_common.create_error_response(401, e)55@app.route('/add_packet_to_user', methods=['POST'])56def restAddPacket():57 try:58 data = rest_common.get_rest_data(request)59 user_service.add_packet_to_user(data)60 return rest_common.create_response(200)61 except InvalidActionException as e:62 return rest_common.create_response(400, e.toDict())63 except SessionElapsedException as e:64 return rest_common.create_error_response(401, e)65 except InvalidSessionIdException as e:66 return rest_common.create_error_response(422, e)67 except PacketNotFoundException as e:68 return rest_common.create_error_response(410, e)69 except PacketAlreadyAddedException as e:70 return rest_common.create_error_response(409, e)71 except CommandFailedException as e:72 return rest_common.create_error_response(504, e)73@app.route('/remove_packet_from_user', methods=['POST'])74def restRemovePacket():75 try:76 data = rest_common.get_rest_data(request)77 user_service.remove_packet_from_user(data)78 return rest_common.create_response(200)79 except NoPacketException as e:80 return rest_common.create_error_response(421, e)81 except (InvalidSessionIdException, NoSessionIdException) as e:82 return rest_common.create_error_response(422, e)83 except InvalidActionException as e:84 return rest_common.create_error_response(400, e)85 except UserUnknownException as e:86 return rest_common.create_error_response(404, e)87 except SessionElapsedException as e:88 return rest_common.create_error_response(401, e)89 except PacketNotFoundException as e:90 return rest_common.create_error_response(410, e)91 except NoSuchPacketAddedException as e:92 return rest_common.create_error_response(428, e)93 except PacketAlreadyAddedException as e:94 return rest_common.create_error_response(409, e)95 96@app.route('/get_packets_from_user/<session_id>', methods=['GET'])97def restGetPacket(session_id):98 try:99 data = {'session_id' : session_id}100 packets = user_service.get_packets_from_user(data)101 if not packets:102 packets = []103 return rest_common.create_response(200, {'packets':packets})104 except InvalidActionException as e:105 return rest_common.create_response(400, e.toDict())106 except SessionElapsedException as e:107 return rest_common.create_error_response(401, e)108 except InvalidSessionIdException as e:109 return rest_common.create_error_response(422, e)110@app.route('/delete_user', methods=['POST'])111def restDeleteUser():112 try:113 data = rest_common.get_rest_data(request)114 user_service.delete_user(data)115 return rest_common.create_response(200)116 except InvalidActionException as e:117 return rest_common.create_response(400, e.toDict())118 except SessionElapsedException as e:119 return rest_common.create_error_response(401, e)120 except InvalidSessionIdException as e:121 return rest_common.create_error_response(422, e)122 except CommandFailedException as e:123 return rest_common.create_error_response(504, e)124@app.route('/logout', methods=['POST'])125def restLogoutUser():126 try:127 data = rest_common.get_rest_data(request)128 user_service.logout_user(data)129 return rest_common.create_response(200)130 except InvalidActionException as e:131 return rest_common.create_response(400, e.toDict())132 except SessionElapsedException as e:133 return rest_common.create_error_response(401, e)134 except InvalidSessionIdException as e:135 return rest_common.create_error_response(422, e)136 except CommandFailedException as e:137 return rest_common.create_error_response(504, e)138def print_help():139 print("Options:\n\t-p Port to use")140if __name__ == '__main__':141 port = 0...
rest-example.py
Source:rest-example.py
...14USERNAME = "admin"15PASSWORD = "admin"16INSECURE = "yes" # ignores certificate checks during REST transactions17# INSECURE = "no"18def get_rest_data(host, uri):19 full_path = "https://" + host + "/" + uri20 21 if INSECURE is "yes":22 # http://urllib3.readthedocs.io/en/latest/advanced-usage.html23 requests.packages.urllib3.disable_warnings()24 return requests.get(full_path, auth=(USERNAME, PASSWORD), verify=False)25 return requests.get(full_path, auth=(USERNAME, PASSWORD))26def get_vs_stats(d):27 # we want the nested dict at key 'entries'28 entries = d['entries']29 30 # We want the nested dict of "nestedStats", which has the REST URL as the key.31 # We could pass that in as a functional argument. Instead, since we know the32 # REST API response has only one key at this level, we'll just iterate.33 for v in entries.values():34 nested_stats = v["nestedStats"]35 # this is the full URL to reach this data, with an optional version parameter36 v_self_link = nested_stats['selfLink']37 38 # The variable 'stats' will contain the nested dict of metrics.39 # Each key maps to a value that contains a dict with one key-value pair.40 # Unfortunately, the key changes for each type of metric. The last value41 # in each nested dict is the one we want.42 #43 # Visually:44 # d['someField']: {"something": 42} }45 # 46 # we want 42, so...47 # blah = d['someField'].values().pop()48 # blah now contains '42'49 stats = nested_stats['entries']50 v_state = stats['status.enabledState'].values().pop()51 v_reason = stats['status.statusReason'].values().pop()52 v_c_bits_in = stats['clientside.bitsIn'].values().pop()53 v_c_bits_out = stats['clientside.bitsOut'].values().pop()54 v_c_pkts_in = stats['clientside.pktsIn'].values().pop()55 v_c_pkts_out = stats['clientside.pktsOut'].values().pop()56 v_c_conn_cur = stats['clientside.curConns'].values().pop()57 v_c_conn_max = stats['clientside.maxConns'].values().pop()58 59 # if you poll a device that hasn't received traffic, you'll have zeros. handle it60 try:61 v_c_pkt_size_in = (v_c_bits_in / 8.0) / v_c_pkts_in62 except ZeroDivisionError:63 v_c_pkt_size_in = 064 try:65 v_c_pkt_size_out = (v_c_bits_out / 8.0) / v_c_pkts_out66 except ZeroDivisionError:67 v_c_pkt_size_out = 068 69 print "From: ", v_self_link70 print "Enabled state: ", v_state71 print "Why: ", v_reason72 73 print "Bits in: ", v_c_bits_in74 print "Bits out: ", v_c_bits_out75 print "Packets in: ", v_c_pkts_in76 print "Packets out: ", v_c_pkts_out77 78 print "Packet size in: ", round(v_c_pkt_size_in, 2), "bytes"79 print "Packet size out:", round(v_c_pkt_size_out, 2), "bytes"80 print "Current conns: ", v_c_conn_cur81 print "Max conns: ", v_c_conn_max82def get_pool_member_health(d):83 print "Pool deb-servers members"84 items = d['items']85 # print type(items)86 for i in items:87 # for k,v in i.items():88 # print k,v89 print "name:", i['name'], "address:", i['address'], "state:", i['state']90def main():91 # query for all configured virtual servers92 resp = get_rest_data(HOST, URI_VS_LIST)93 j = resp.json()94 # acquire virtual server names95 vs_names = []96 for k in j['items']:97 for ksub in k:98 if ksub == 'name':99 vs_names.append(k['name'])100 #print k['name']101 # print statistics for each virtual server102 # note: each VS will be a separate REST API query. rate limit at scale, as needed.103 for item in vs_names:104 uri_query = URI_VS_LIST + item + STATS_LOC105 resp = get_rest_data(HOST, uri_query)106 j = resp.json()107 print "virtual server: ", item # vs name108 get_vs_stats(j)109 print # trailing newline110 111 # quick health information for every pool112 resp = get_rest_data(HOST, URI_POOL)113 j = resp.json()114 get_pool_member_health(j)115 print116if __name__ == "__main__":...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run yandex-tank automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
