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.