Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use allocate_hosts method in localstack

Best Python code snippet using localstack_python

ufm_logical_servers.py

Source:ufm_logical_servers.py

...293            Logger.log_message(response)294            return295        Logger.log_message(response.text, LOG_LEVELS.ERROR)296    @staticmethod297    def auto_allocate_hosts(env_name, ls_name, number_of_hosts):298        url = f'{UfmLogicalServersConstants.ENV_LS_API_URL.format(env_name)}' \299              f'/{ls_name}/auto_assign-computes'300        payload = {301            "total_computes": number_of_hosts302        }303        response = ufm_rest_client.send_request(url, HTTPMethods.PUT, payload)304        if response and response.status_code == HTTPStatus.OK:305            Logger.log_message(f'{number_of_hosts} hosts allocated successfully to the logical server: {ls_name}'306                               f' under the environment: {env_name}')307            Logger.log_message(response.json())308            return True309        Logger.log_message(response.text, LOG_LEVELS.ERROR)310    @staticmethod311    def allocate_hosts(env_name, ls_name, guids_list):312        url = f'{UfmLogicalServersConstants.ENV_LS_API_URL.format(env_name)}' \313              f'/{ls_name}/allocate-computes'314        payload = {315            "computes": guids_list316        }317        response = ufm_rest_client.send_request(url, HTTPMethods.PUT, payload)318        if response and response.status_code == HTTPStatus.OK:319            Logger.log_message(f'{guids_list} allocated successfully to the logical server: {ls_name} '320                               f'under the environment: {env_name}')321            Logger.log_message(response.json())322            return True323        Logger.log_message(response.text, LOG_LEVELS.ERROR)324    @staticmethod325    def add_network_interfaces(env_name, ls_name, networks_list):326        url = f'{UfmLogicalServersConstants.ENV_LS_API_URL.format(env_name)}' \327              f'/{ls_name}/network_interfaces'328        payload = []329        for network in networks_list:330            payload.append({"network": network})331        response = ufm_rest_client.send_request(url, HTTPMethods.POST, payload)332        if response and response.status_code == HTTPStatus.OK:333            Logger.log_message(f'{networks_list} networks interfaces are added successfully '334                               f'to the logical server: {ls_name} '335                               f'under the environment: {env_name}')336            Logger.log_message(response.json())337            return True338        Logger.log_message(response.text, LOG_LEVELS.ERROR)339if __name__ == "__main__":340    # init app args341    args = ArgsParser.parse_args("UFM Logical Servers Management", UfmLogicalServersConstants.args_list)342    # init app config parser & load config files343    config_parser = UfmLsConfigParser(args)344    # init logs configs345    logs_file_name = config_parser.get_logs_file_name()346    logs_level = config_parser.get_logs_level()347    Logger.init_logs_config(logs_file_name, logs_level)348    # init ufm rest client349    ufm_rest_client = UfmRestClient(host = config_parser.get_ufm_host(),350                                    client_token=config_parser.get_ufm_access_token())351    if config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("get_envs")):352        UfmLsManagement.get_envs()353    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("get_env")):354        env_name = config_parser.get_config_value(config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("get_env")),355                                                  None,None)356        UfmLsManagement.get_envs(env_name)357    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("create_env")):358        try:359            env_name = config_parser.get_name()360        except ValueError as e:361            Logger.log_missing_args_message(UfmLogicalServersConstants.LS_OPERATIONS.get("create_env"),362                                            UfmLogicalServersConstants.API_NAME)363            sys.exit(1)364        desc = config_parser.get_description()365        UfmLsManagement.create_env(env_name, desc)366    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("delete_env")):367        env_name = config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("delete_env"))368        UfmLsManagement.delete_env(env_name)369    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("get_networks")):370        UfmLsManagement.get_networks()371    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("get_network")):372        network_name = config_parser.get_config_value(config_parser.args_dict.get(373            UfmLogicalServersConstants.LS_OPERATIONS.get("get_network")),None,None)374        UfmLsManagement.get_networks(network_name)375    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("delete_network")):376        network_name = config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("delete_network"))377        UfmLsManagement.delete_network(network_name)378    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("create_network")):379        try:380            name = config_parser.get_name()381        except ValueError as e:382            Logger.log_missing_args_message(UfmLogicalServersConstants.LS_OPERATIONS.get("create_network"),383                                            UfmLogicalServersConstants.API_NAME)384            sys.exit(1)385        description = config_parser.get_description()386        pkey = config_parser.get_pkey()387        UfmLsManagement.create_network(name=name, desc=description, pkey=pkey)388    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("get_logical_servers")):389        UfmLsManagement.get_ls()390    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("delete_logical_server")):391        try:392            ls_name = config_parser.get_name()393            env_name = config_parser.get_env()394        except ValueError as e:395            Logger.log_missing_args_message(UfmLogicalServersConstants.LS_OPERATIONS.get("delete_logical_server"),396                                            UfmLogicalServersConstants.API_NAME,397                                            UfmLogicalServersConstants.API_ENV)398            sys.exit(1)399        UfmLsManagement.delete_ls(env_name, ls_name)400    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("create_logical_server")):401        try:402            ls_name = config_parser.get_name()403            env_name = config_parser.get_env()404        except ValueError as e:405            Logger.log_missing_args_message(UfmLogicalServersConstants.LS_OPERATIONS.get("create_logical_server"),406                                            UfmLogicalServersConstants.API_NAME,407                                            UfmLogicalServersConstants.API_ENV)408            sys.exit(1)409        description = config_parser.get_description()410        UfmLsManagement.create_ls(env_name, ls_name, desc=description)411    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("get_free_hosts")):412        UfmLsManagement.get_free_hosts()413    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("auto_allocate_hosts")):414        try:415            ls_name = config_parser.get_name()416            env_name = config_parser.get_env()417            total_computes = config_parser.get_total_computes()418        except ValueError as e:419            Logger.log_missing_args_message(UfmLogicalServersConstants.LS_OPERATIONS.get("auto_allocate_hosts"),420                                            UfmLogicalServersConstants.API_NAME,421                                            UfmLogicalServersConstants.API_ENV,422                                            UfmLogicalServersConstants.API_TOTAL_COMPUTES)423            sys.exit(1)424        UfmLsManagement.auto_allocate_hosts(env_name, ls_name, total_computes)425    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("allocate_hosts")):426        try:427            ls_name = config_parser.get_name()428            env_name = config_parser.get_env()429            guids = config_parser.get_computes()430        except ValueError as e:431            Logger.log_missing_args_message(UfmLogicalServersConstants.LS_OPERATIONS.get("allocate_hosts"),432                                            UfmLogicalServersConstants.API_NAME,433                                            UfmLogicalServersConstants.API_ENV,434                                            UfmLogicalServersConstants.API_COMPUTES)435            sys.exit(1)436        UfmLsManagement.allocate_hosts(env_name, ls_name, guids)437    elif config_parser.args_dict.get(UfmLogicalServersConstants.LS_OPERATIONS.get("add_network_interfaces")):438        try:439            ls_name = config_parser.get_name()440            env_name = config_parser.get_env()441            networks = config_parser.get_networks()442        except ValueError as e:443            Logger.log_missing_args_message(UfmLogicalServersConstants.LS_OPERATIONS.get("add_network_interfaces"),444                                            UfmLogicalServersConstants.API_NAME,445                                            UfmLogicalServersConstants.API_ENV,446                                            UfmLogicalServersConstants.API_NETWORKS)447            sys.exit(1)448        UfmLsManagement.add_network_interfaces(env_name, ls_name, networks)449    else:450        message = "You must provide one of the following operations: "+ \...

Perf_Info_menu_V3.py

Source:Perf_Info_menu_V3.py

1#!/usr/bin/python -tt23import getopt4import subprocess5import sys6import os78menu_actions = {}910def ssh_cmd(target,cmd):11	full_cmd = 'ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o BatchMode=yes root@%s %s' %(target,cmd)12	cmd_return = subprocess.Popen(full_cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)13	return cmd_return14	15def hosts_info(list="all",sort="name"):16	#print sys.argv[1]17	if list == "all":18		fccon_output = ssh_cmd('stcon.xiv.ibm.com','/opt/FCCon/fcconnect.pl -dbop showhosts|grep "Free\|Used"')19	else:20		fccon_output = ssh_cmd('stcon.xiv.ibm.com','/opt/FCCon/fcconnect.pl -dbop showhosts |grep Free')21	fccon_output.wait()22	host_list=[]23	host_list_noping=[]24	host_list_nossh=[]25	host_list_ok={}26	# creating the required host list based on DB27	for line in fccon_output.stdout.readlines():28		#host_list.append(line.split(",")[0])29		host=line.split(",")[0]30		host_os = line.split(",")[1]31	#for host in host_list:32		# checking hosts availability33#		host=host+".eng.rtca"34		ping_cmd = 'ping -c1 %s' %(host)35		ping = subprocess.Popen(ping_cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)36		if ping.wait() !=0:37			#print "failed to ping host: %s" %(host)38			host_list_noping.append(host)39		else:40			#checking ssh connectivity41			ssh = ssh_cmd(host,'date')42			if ssh.wait() !=0:43				host_list_nossh.append(host)44			else:45				if host_os == 'linux':46					# getting Linux os version47					#os_version = os.popen("ssh root@%s cat /etc/redhat-release | cut -f 7 -d " ") %(host)48					os_version = ssh_cmd(host,'cat /etc/redhat-release | cut -f 7 -d " "').stdout.readline(3)49					#getting Java version50					java_version="na"51					java_version = ssh_cmd(host,'java -version 2>&1 | awk "/version/{print $NF}"').stdout.readline().rstrip('\n')52					#if java_version =='':53					#	java_version = "na"					54					# getting FC HBA info55					total_active = 056					fc_speed = "na"57					fc_port_list = ssh_cmd(host,'ls /sys/class/fc_host/').stdout.readlines()58					total_ports = len(fc_port_list)59					for port in fc_port_list:60						#check status61						port_status = ssh_cmd(host,'cat /sys/class/fc_host/%s/port_state' %(port.rstrip('\n')))62						if port_status.wait() !=0:63							print "failed to detect fc status for %s" %(host)64						else:65							if port_status.stdout.readline().rstrip('\n') == 'Online':66								total_active = total_active + 167								#check speed68								port_speed = ssh_cmd(host,'cat /sys/class/fc_host/%s/speed' %(port.rstrip('\n')))69								if port_speed.wait()!=0:70									print "failed to detect fc speed for %s" %(host)71								else:72									fc_speed = port_speed.stdout.readline().rstrip('\n')73					if fc_speed.rsplit()[0].isdigit():74							host_bw = total_active*int(fc_speed.rsplit()[0])*10075					else:76						host_bw = 'na'77					total_cpu=ssh_cmd(host,'cat /proc/cpuinfo |grep processor|wc -l').stdout.readline().rstrip('\n')78					total_memory=ssh_cmd(host,'cat /proc/meminfo |grep MemTotal|awk "{print $2$3}"').stdout.readline().rstrip('\n')79					host_list_ok[host] = {'OS':host_os,'version':os_version,'java version':java_version,'fc_speed':fc_speed,'total_fc_ports':total_ports,'total_fc_active':total_active,'host_bw_MB/s':host_bw,'total_cpu':total_cpu,'total_memory':total_memory}80				elif host_os == 'esx':81					print "host %s is esxi" %(host)82					host_list_ok[host] = 'esxi'					83	# need to edit print to look nice84	if sort == 'name':85		print "list of free hosts sorted by %s:" %(sort)86		for host_key in sorted(host_list_ok):87			print host_key,host_list_ok.get(host_key)88	else: 89		sort_dir={}90		print "list of free hosts sorted by %s:" %(sort)91		for host_key in host_list_ok:92			sort_dir[host_key] = host_list_ok.get(host_key).get(sort)93		for key, value in sorted(sort_dir.iteritems(), key=lambda (k,v): (v,k)):94			print "host name:%s %s:%s" %(key, sort, value)95	#print free_host_list_ok96	print "list of no ping hosts:"97	print host_list_noping98	print "list of no ssh hosts:"99	print host_list_nossh100	return101			102# Define a main() function that prints a little greeting.103def main():104  # Aks the usr to select an option from the menu105  print "Please select your option:"106  print "1: List hosts info"107  print "2: List storage info"108  print "3: Allocate desired hosts"109  print "b: back"110  print "x: Exit"111  option = raw_input("Enter your option >>> ")112  exec_main(option)113114#Execute selected menu  115def exec_main(option):116#    os.system('clear')117    selected_option = option.lower()118    if selected_option == '':119      print "No option was selected"120      menu_actions['main_menu']()121    else:122        try:123          menu_actions[selected_option]()124        except KeyError:125          print "Invalid selection, please try again.\n"126          menu_actions['main_menu']()127    return128129# Back to main menu130def back():131  menu_actions['main_menu']()132 133# Exit program134def exit():135  sys.exit()136137# host listing menus 138def list_hosts():139  print "\n"140  print "Please select your host list option:"141  print "1: List all hosts info"142  print "2: List Free hosts info"143  print "3: List hosts with no ping"144  print "4: List hosts with no ssh access"145  print "b: back"146  print "x: Exit"147  option = raw_input("Enter your option >>> ")148  menu=list_menu149  list_opt=exec_menu(menu,option)150  print "\n"151  print "Please select your host sort option:"152  print "1: Sort by host name"153  print "2: Sort by host FC speed"154  print "3: Sort by host total FC active"155  print "4: Sort by host total bw"156  print "b: back"157  print "x: Exit"158  option = raw_input("Enter your option >>> ")159  menu = sort_menu160  sort_opt=exec_menu(menu,option)161  hosts_info(list=list_opt,sort=sort_opt)162  print "\n"163  #print "b. Back"164  #print "x. Quit"165  #option = raw_input(" >>>  ")166  #exec_menu(option)167  return menu_actions['main_menu']()168 169def exec_menu(menu,option):170    selected_option = option.lower()171    if selected_option == '':172      print "No option was selected"173      menu['main_menu']()174    else:175        try:176		if selected_option.isdigit():177			return menu[selected_option]178		else:179			menu[selected_option]()180        except KeyError:181          print "Invalid selection, please try again.\n"182          menu['main_menu']()183    return 184 185# Menu definition186menu_actions = {187    'main_menu': main,188    '1': list_hosts,189    '2': "list_storage",  #to be developed190	'3': "allocate_hosts", #to be developed191    'b': back,192    'x': exit,193}194195list_menu = {196    'main_menu': main,197    '1': "all",198    '2': "free",199	'3': "noping",200	'4': "nossh",201    'b': back,202    'x': exit,203}  204205sort_menu = {206    'main_menu': main,207    '1': "name",208    '2': "fc_speed",209    '3': "total_fc_active",210    '4': "host_bw_MB/s",211    'b': back,212    'x': exit,213}  214215# This is the standard boilerplate that calls the main() function.216if __name__ == '__main__':
...

ipythonsession.py

Source:ipythonsession.py

1# coding: utf-82import boto33# establishing session with profile4session = boto3.Session(profile_name='pythonAutomation')5# establishing session with s36s3 = session.resource('s3')7# iterate through buckets8for bucket in s3.buckets.all():9    print(bucket)10    11# establishing session with ec212ec2_client = session.client('ec2')13#ec2_client.allocate_address...

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.