Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use test_network_basic_ops method in tempest
Best Python code snippet using tempest_python
run_tests.py
Source:run_tests.py 
1#!/usr/bin/python2import click3import datetime4import logging as log5import os6import paramiko7import subprocess8import sys9import time10import uuid11from cleanup_tempest import cleanup_tempest12import tempest_conf as cfg_template13from fabric.api import sudo14JUJU='juju'15DIRECTOR='director'16JUJU_RC='admin-openrc.sh'17DIRECTOR_RC='overcloudrc'18PPATH='export PYTHONPATH=/home/noiro/noirotest && '19CMD_TIME_WAIT = 1520# Wait 3 hours max21TEMPEST_WAIT_TIME = (60*60*3)22NOIRO_SETUP_WAIT_TIME = (60*2)23ML2_SANITY_WAIT_TIME = (60*30)24GBP_SANITY_WAIT_TIME = (60*30)25EAST_WEST_WAIT_TIME = (60*45)26NORTH_SOUTH_WAIT_TIME = (60*60)27CLEAR_IP_TABLES_RULE='iptables -D INPUT -j REJECT --reject-with icmp-host-prohibited'28default_password = 'noir0123'29TESTR_DOT_CONF="""30[DEFAULT]31test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} \32 OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} \33 OS_TEST_TIMEOUT=${OS_TEST_TIMEOUT:-500} \34 ${PYTHON:-python} -m subunit.run discover -t /usr/local/lib/python2.7/dist-packages/tempest /usr/local/lib/python2.7/dist-packages/tempest/test_discover $LISTOPT $IDOPTION35test_id_option=--load-list $IDFILE36test_list_option=--list37group_regex=([^\.]*\.)*38"""39# Dictionary that shows which tempest tests to exclude, by release40excluded_test_matrix = {41 'stable/newton': [42 ('tempest.api.network.test_floating_ips.FloatingIPTestJSON.test_floating_ip_update_different_router',43 'Test uses overlapping CIDRs in the same VRF'),44 ('tempest.api.network.test_networks.NetworksTest.test_external_network_visibility',45 'BUG: https://github.com/noironetworks/support/issues/916'),46 ('tempest.api.network.test_security_groups.SecGroupTest.test_create_show_delete_security_group_rule',47 'BUG: https://github.com/noironetworks/support/issues/710'),48 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_subnet_details',49 "BIG: https://github.com/noironetworks/support/issues/915"),50 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_update_router_admin_state',51 'BUG: https://github.com/noironetworks/support/issues/491'),52 ('neutron.tests.tempest.api.test_floating_ips_negative.FloatingIPNegativeTestJSON.test_associate_floatingip_with_port_with_floatingip',53 'Test uses overlapping CIDRs in the same VRF'),54 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sg_with_timestamp',55 'BUG: https://github.com/noironetworks/support/issues/710'),56 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sgrule_with_timestamp',57 'BUG: https://github.com/noironetworks/support/issues/710'),58 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sg_attribute_with_timestamp',59 'BUG: https://github.com/noironetworks/support/issues/710'),60 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sgrule_attribute_with_timestamp',61 'BUG: https://github.com/noironetworks/support/issues/710'),62 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_update_sg_with_timestamp',63 'BUG: https://github.com/noironetworks/support/issues/710')64 ],65 'stable/ocata': [66 ('tempest.api.network.test_floating_ips.FloatingIPTestJSON.test_floating_ip_update_different_router',67 'Test uses overlapping CIDRs in the same VRF'),68 ('tempest.api.network.test_networks.NetworksTest.test_external_network_visibility',69 'BUG: https://github.com/noironetworks/support/issues/916'),70 ('tempest.api.network.test_security_groups.SecGroupTest.test_create_show_delete_security_group_rule',71 'BUG: https://github.com/noironetworks/support/issues/710'),72 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_subnet_details',73 "BIG: https://github.com/noironetworks/support/issues/915"),74 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_update_router_admin_state',75 'BUG: https://github.com/noironetworks/support/issues/491'),76 ('neutron.tests.tempest.api.test_floating_ips_negative.FloatingIPNegativeTestJSON.test_associate_floatingip_with_port_with_floatingip',77 'Test uses overlapping CIDRs in the same VRF'),78 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sg_with_timestamp',79 'BUG: https://github.com/noironetworks/support/issues/710'),80 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sgrule_with_timestamp',81 'BUG: https://github.com/noironetworks/support/issues/710'),82 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sg_attribute_with_timestamp',83 'BUG: https://github.com/noironetworks/support/issues/710'),84 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sgrule_attribute_with_timestamp',85 'BUG: https://github.com/noironetworks/support/issues/710'),86 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_update_sg_with_timestamp',87 'BUG: https://github.com/noironetworks/support/issues/710')88 ],89 'stable/pike': [90 ('tempest.api.network.test_floating_ips.FloatingIPTestJSON.test_floating_ip_update_different_router',91 'Test uses overlapping CIDRs in the same VRF'),92 ('tempest.api.network.test_networks.NetworksTest.test_external_network_visibility',93 'BUG: https://github.com/noironetworks/support/issues/916'),94 ('tempest.api.network.test_security_groups.SecGroupTest.test_create_show_delete_security_group_rule',95 'BUG: https://github.com/noironetworks/support/issues/710'),96 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_subnet_details',97 "BIG: https://github.com/noironetworks/support/issues/915"),98 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_update_router_admin_state',99 'BUG: https://github.com/noironetworks/support/issues/491'),100 ('neutron.tests.tempest.api.test_floating_ips_negative.FloatingIPNegativeTestJSON.test_associate_floatingip_with_port_with_floatingip',101 'Test uses overlapping CIDRs in the same VRF'),102 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sg_with_timestamp',103 'BUG: https://github.com/noironetworks/support/issues/710'),104 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sgrule_with_timestamp',105 'BUG: https://github.com/noironetworks/support/issues/710'),106 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sg_attribute_with_timestamp',107 'BUG: https://github.com/noironetworks/support/issues/710'),108 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sgrule_attribute_with_timestamp',109 'BUG: https://github.com/noironetworks/support/issues/710'),110 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_update_sg_with_timestamp',111 'BUG: https://github.com/noironetworks/support/issues/710')112 ],113 'stable/queens': [114 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_subnet_details',115 "BIG: https://github.com/noironetworks/support/issues/915"),116 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_update_router_admin_state',117 'BUG: https://github.com/noironetworks/support/issues/491'),118 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_port_security_macspoofing_port',119 'Why is this failing?'),120 ('tempest.api.network.test_floating_ips.FloatingIPTestJSON.test_floating_ip_update_different_router',121 'Test uses overlapping CIDRs in the same VRF'),122 ('tempest.api.compute.servers.test_attach_interfaces.AttachInterfacesTestJSON.test_create_list_show_delete_interfaces_by_fixed_ip',123 'Why is this test failing?')124 ],125}126# A lot of this stuff may end up being throwaway code -- this should127# live in the pytest framework. However, I need something I can use128# until we've migrated the tests, which is why this code exists.129class Runner(object):130 """Base runner class.131 This class is used to run a family of commands for a given132 framework.133 """134 def __init__(self):135 # TODO: divine this value from config136 self.plugin_type = 'aim'137 self.undercloud_type = DIRECTOR138 self.KEY = 'source ~/' + DIRECTOR_RC + ' && '139 self.ssh_clients = {}140 localtime = datetime.datetime.now()141 env_params = {'y': datetime.datetime.now().year,142 'm': datetime.datetime.now().month,143 'd': datetime.datetime.now().day,144 'h': datetime.datetime.now().hour,145 'i': datetime.datetime.now().minute,146 's': datetime.datetime.now().second}147 self.env_name = "%(y)s_%(m)s_%(d)s_%(h)s_%(i)s_%(s)s" % env_params148 def get_ssh_client(self, host, username='root'):149 ssh_client = self.ssh_clients.get(host)150 if not ssh_client:151 ssh_client = paramiko.SSHClient()152 ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())153 ssh_client.load_system_host_keys()154 print "connecting to " + host155 ssh_client.connect(host, username=username)156 self.ssh_clients[host] = ssh_client157 return ssh_client158 def remote_cmd(self, ssh_client, cmd, wait=CMD_TIME_WAIT):159 ssh_stdin, ssh_stdout, ssh_stderr = ssh_client.exec_command(cmd)160 if wait:161 endtime = time.time() + wait162 while not ssh_stdout.channel.eof_received:163 time.sleep(1)164 if time.time() > endtime:165 ssh_stdout.channel.close()166 return []167 return ssh_stdout.read(), ssh_stderr.read()168 def setup_passwordless_ssh(self, user='root', hosts=None):169 for host in hosts:170 cmd ='ssh-keygen -f "/home/noiro/.ssh/known_hosts" -R %s' % host171 print cmd172 subprocess.check_output(['bash','-c', cmd])173 cmd1 = 'ssh -o "StrictHostKeyChecking no" %(user)s@%(host)s ' % {'user': user, 'host': host}174 cmd2 = "'ls -l'"175 cmd = cmd1 + cmd2176 print cmd177 subprocess.check_output(['bash','-c', cmd])178 cmd = "cat ~/.ssh/id_rsa.pub | ssh %(user)s@%(host)s 'cat >> .ssh/authorized_keys'" % {'user': user, 'host': host}179 print cmd180 subprocess.check_output(['bash','-c', cmd])181 subprocess.check_output(['bash','-c', cmd])182 def configure_setup_for_test(self):183 pass184 def run_test(self, test_name):185 pass186 def run_all_tests(self):187 pass188 def cleanup(self):189 pass190class TempestTestRunner(Runner):191 def __init__(self, controller = None,192 compute1 = None, compute2 = None, ext_rtr = None,193 undercloud_type=DIRECTOR):194 super(TempestTestRunner, self).__init__()195 self.controller_host = controller196 self.compute1_host = compute1197 self.compute2_host = compute2198 self.external_router = ext_rtr199 self.undercloud_type = undercloud_type200 if undercloud_type == DIRECTOR:201 self.cli_host = self.controller_host202 self.cli_user = 'heat-admin'203 self.KEY = 'source ~/' + DIRECTOR_RC + ' && '204 elif undercloud_type == JUJU:205 self.cli_host = self.external_router206 self.cli_user = 'noiro'207 self.KEY = 'source ~/' + JUJU_RC + ' && '208 else:209 print "Unsupported undercloud type: " + undercloud_type210 self.hosts = [self.external_router, self.controller_host,211 self.compute1_host, self.compute2_host]212 self.image_uuid = None213 self.flavor_uuid = None214 self.alt_image_uuid = None215 self.alt_flavor_uuid = None216 self.network_uuid = None217 self.version = self.get_openstack_version()218 def get_openstack_version(self):219 # we rely on the openstack client repo220 cmd1 = "cd python-openstackclient/ && "221 cmd2 = "git status | grep 'branch stable\|eol' "222 cmd3 = "| awk 'NF>1{print $NF}' && cd .."223 cmd = self.KEY + cmd1 + cmd2 + cmd3224 print cmd225 subprocess.check_output(['bash','-c', cmd])226 version_string = subprocess.check_output(['bash','-c', cmd])227 version_list = [version.strip()228 for version in version_string.split("\n") if version]229 version = version_list[0]230 if version == 'newton-eol':231 version = 'stable/newton'232 return version233 234 def configure_setup_for_test(self):235 images = [{'image': 'cirros-0.3.5-x86_64-disk.img',236 'name': 'cirros.new'},237 {'image': 'cirros-0.3.5-x86_64-disk.img',238 'name': 'cirros.alt'},239 {'image': 'ubuntu_multi_nics.qcow2',240 'name': 'ubuntu_multi_nics'}]241 for image in images:242 image_uuid = self.upload_test_image(image_file=image['image'],243 image_name=image['name'])244 if image['name'] is 'cirros.new':245 self.image_uuid = image_uuid246 if image['name'] is 'cirros.alt':247 self.alt_image_uuid = image_uuid248 # for OSD and JuJu installs, we also have to configure the image flavor249 print("undercloud type is %s" % self.undercloud_type)250 if self.undercloud_type in [DIRECTOR, JUJU]:251 print("configuring flavor UUID")252 flavor_list = [{'name': 'm1.tiny', 'cpus': '1', 'ram': '512',253 'disk': '1', 'swap': '0'},254 {'name': 'm1.alt_tiny', 'cpus': '1', 'ram': '512',255 'disk': '1', 'swap': '0'},256 {'name': 'm1.medium', 'cpus': '2', 'ram': '4096',257 'disk': '40', 'swap': '0'},258 {'name': 'm1.large', 'cpus': '4', 'ram': '8192',259 'disk': '80', 'swap': '0'}]260 for flavor in flavor_list:261 flavor_uuid = self.configure_image_flavor(flavor)262 if flavor['name'] is 'm1.tiny':263 self.flavor_uuid = flavor_uuid264 if flavor['name'] is 'm1.alt_tiny':265 self.alt_flavor_uuid = flavor_uuid266 self.configure_neutron()267 self.network_uuid = self.create_external_network(self.cli_host, username=self.cli_user)268 self.create_tempest_config()269 #self.exclude_tests()270 def get_test_info(self, test):271 test_pieces = test.split(".")272 path = test_pieces[0] + "/" + "/".join(test_pieces[0:-2]) + ".py"273 test_class = test_pieces[-2:-1][0]274 test_name = test.split(".")[-1:][0]275 return path, test_class, test_name276 def exclude_tests(self):277 for test, reason in excluded_test_matrix[self.version]:278 path, test_class, test_name = self.get_test_info(test)279 def add_line(file_name, before_line, new_line):280 # find the first instance of the string281 index = None282 all_lines = None283 with open(file_name, "r") as fd:284 all_lines = fd.readlines()285 for count in range(len(all_lines)):286 line = all_lines[count]287 if before_line in line:288 strip_line = line[:-1]289 indent_len = len(strip_line) - len(strip_line.strip())290 indent = line[:indent_len] if indent_len else ''291 new_line = indent + new_line292 index = count293 break294 if index:295 all_lines.insert(index, new_line)296 with open(file_name, "w+") as fd:297 contents = "".join(all_lines)298 fd.write(contents)299 300 cmd = "egrep testtools %s" % path301 try:302 subprocess.check_output(['bash','-c', cmd])303 except Exception as e:304 # not present, so add testtools305 add_line(path, "import", "import testtools\n")306 pass307 # need to escape backslashes308 309 reason_str = '@testtools.skip("' + reason + '")\n'310 # Skip this test311 add_line(path, 'def ' + test_name, reason_str)312 def set_ext_rtr_routes(self):313 ssh_client = self.get_ssh_client(self.external_router, username='noiro')314 cmd = ''315 def upload_test_image(self, image_file='cirros-0.3.5-x86_64-disk.img',316 image_name='cirros.new'):317 """Upload a test image to the OpenStack controller318 This uploads a user-defined image to glance on the OpenStack319 controller. It presumes that the image is locally accessible to320 the machine running this script, that the machine has the321 OpenStack python client library installed, and that the proper322 credentials are setup in the environment (e.g. source .openrc).323 """324 cmd = self.KEY + 'openstack image show %s -c id -f value' % image_name325 try:326 image_uuid = subprocess.check_output(['bash','-c', cmd])327 # Make sure it's a valid UUID328 try:329 image_uuid = [iuuid.strip() for iuuid in image_uuid.split("\n") if iuuid][0]330 valid_uuid = uuid.UUID(image_uuid)331 # network exists - return UUID332 print(image_uuid)333 return image_uuid334 except ValueError:335 pass336 except:337 pass338 cmd1 = "openstack image create --public"339 cmd2 = " --disk-format %s" % 'qcow2'340 cmd3 = " --container-format %s" % 'bare'341 cmd4 = " --file %s " % image_file342 cmd = self.KEY + cmd1 + cmd2 + cmd3 + cmd4 + image_name343 print cmd344 subprocess.check_output(['bash','-c', cmd])345 cmd = self.KEY + 'openstack image show %s -c id -f value' % image_name346 image_uuid = subprocess.check_output(['bash','-c', cmd])347 image_uuid = [iuuid.strip() for iuuid in image_uuid.split("\n") if iuuid][0]348 print(image_uuid)349 return image_uuid350 def configure_image_flavor(self, flavor=None):351 """Configure image flavor for tempest tests.352 Configure a flavor used for tempest testing.353 """354 if not flavor:355 return356 print "configuring image flavor"357 cmd = self.KEY + 'openstack flavor show %s -c id -f value' % flavor['name']358 print cmd359 try:360 flavor_uuid = subprocess.check_output(['bash','-c', cmd])361 # Make sure it's a valid UUID362 try:363 flavor_uuid = [iuuid.strip() for iuuid in flavor_uuid.split("\n") if iuuid][0]364 valid_uuid = uuid.UUID(flavor_uuid)365 # network exists - return UUID366 print(flavor_uuid)367 return flavor_uuid368 except ValueError:369 pass370 except:371 pass372 cmd1 = "openstack flavor create --ram %s" % flavor['ram']373 cmd2 = " --disk %s" % flavor['disk']374 cmd3 = " --vcpus %s" % flavor['cpus']375 cmd4 = " --swap %s " % flavor['swap']376 cmd5 = " --public "377 cmd = self.KEY + cmd1 + cmd2 + cmd3 + cmd4 + cmd5 + flavor['name']378 print cmd379 subprocess.check_output(['bash','-c', cmd])380 cmd = self.KEY + 'openstack flavor show %s -c id -f value' % flavor['name']381 flavor_uuid = subprocess.check_output(['bash','-c', cmd])382 flavor_uuid = [iuuid.strip() for iuuid in flavor_uuid.split("\n") if iuuid][0]383 print(flavor_uuid)384 return flavor_uuid385 def configure_neutron(self):386 """Set up neutron for tempest testing.387 This configures neutron so that tempest tests can388 be run.389 """390 pass391 def create_external_network(self, host, username='heat-admin'):392 """Create an external network for tempest testing.393 Tempest requires access to instances via a public network.394 This creates a public network that is used for tempest395 testing on noiro fabrics.396 """397 ssh_client = self.get_ssh_client(host, username=username)398 cmd = self.KEY + 'openstack network show sauto_l3out-2 -c id -f value'399 net_uuid, stderr = self.remote_cmd(ssh_client, cmd)400 if not stderr:401 net_uuid = [nuuid.strip() for nuuid in net_uuid.split("\n") if nuuid][0]402 # Make sure it's a valid UUID403 try:404 valid_uuid = uuid.UUID(net_uuid)405 # network exists - return UUID406 print(net_uuid)407 return net_uuid408 except ValueError:409 pass410 # Doesn't exist -- create it411 cmd1 = "neutron net-create sauto_l3out-2 --router:external True "412 cmd2 = "--shared --apic:distinguished_names type=dict "413 cmd3 = "ExternalNetwork=uni/tn-common/out-sauto_l3out-2/instP-sauto_l3out-2_epg"414 cmd = self.KEY + cmd1 + cmd2 + cmd3415 print cmd416 self.remote_cmd(ssh_client, cmd)417 cmd1 = "neutron subnet-create sauto_l3out-2 60.60.60.0/24 "418 cmd2 = "--name ext-subnet --disable-dhcp --gateway 60.60.60.1"419 cmd = self.KEY + cmd1 + cmd2420 self.remote_cmd(ssh_client, cmd)421 if self.version != 'stable/newton':422 cmd1 = "neutron subnet-create sauto_l3out-2 66.66.66.0/24 "423 cmd2 = "--name snat-subnet --gateway 66.66.66.1 --apic:snat_host_pool True"424 cmd = self.KEY + cmd1 + cmd2425 self.remote_cmd(ssh_client, cmd)426 ssh_client = self.get_ssh_client(host)427 cmd = self.KEY + 'openstack network show sauto_l3out-2 -c id -f value'428 net_uuid, _ = self.remote_cmd(ssh_client, cmd)429 print(net_uuid)430 print(_)431 net_uuid = [nuuid.strip() for nuuid in net_uuid.split("\n") if nuuid][0]432 print(net_uuid)433 return net_uuid434 def create_tempest_config(self):435 ssh_client = self.get_ssh_client(self.external_router, username='noiro')436 cmd = 'tempest init %s' % self.env_name437 print cmd438 self.remote_cmd(ssh_client, cmd)439 # Get the IP address of the OpenStack controller440 cmd = self.KEY + "echo $OS_AUTH_URL | awk -F'/' '{print $3}' | awk -F':' '{print $1}'"441 controller_ip, _ = self.remote_cmd(ssh_client, cmd)442 controller_ip = [ip.strip() for ip in controller_ip.split("\n") if ip][0]443 tempest_params = {}444 tempest_params.update({'cwd': os.getcwd() + '/' + self.env_name,445 'image_uuid': self.image_uuid,446 'flavor_uuid': self.flavor_uuid,447 'alt_image_uuid': self.alt_image_uuid,448 'alt_flavor_uuid': self.alt_flavor_uuid,449 'controller_ip': controller_ip,450 'controller_password': 'noir0123',451 'external_network': self.network_uuid})452 tempest_cfg = [cfg % tempest_params for cfg in cfg_template.ALL_TEMPLATES]453 tempest_conf_file = '%s/etc/tempest.conf' % self.env_name454 cfg_file = open(tempest_conf_file,'w') 455 for cfg in tempest_cfg:456 cfg_file.write(cfg)457 cfg_file.close()458 testr_dot_conf_file = '%s/.testr.conf' % self.env_name459 dot_conf_file = open(testr_dot_conf_file,'w') 460 for cfg in TESTR_DOT_CONF:461 dot_conf_file.write(cfg)462 dot_conf_file.close()463 def run_test(self, test_name):464 cmd = 'ostestr --regex %s' % test_name465 subprocess.check_output(['bash','-c', cmd])466 def run_all_tests(self, host):467 # Create a tempest environment, using the local date/time468 ssh_client = self.get_ssh_client(host)469 cmd = 'cd %s && tempest run' % self.env_name470 self.remote_cmd(ssh_client, cmd, wait=TEMPEST_WAIT_TIME)471 #subprocess.check_output(['bash','-c', cmd])472 def cleanup(self):473 ssh_client = self.get_ssh_client(self.controller_host)474 cleanup_tempest(self.plugin_type, ssh_client)475class NoiroTestRunner(Runner):476 def __init__(self, ext_rtr=None):477 super(NoiroTestRunner, self).__init__()478 self.external_router = ext_rtr479 self.ssh_client = self.get_ssh_client(ext_rtr, username='noiro')480 def configure_setup_for_test(self):481 # Run the noirotest setup482 cmd = 'cd noirotest && python setup.py'483 self.remote_cmd(self.ssh_client, cmd, wait=NOIRO_SETUP_WAIT_TIME)484 def save_results(self, test, out, err):485 results_filename = '%(env)s_%(test)s_results' % {'env': self.env_name,486 'test': test}487 results_file = open(results_filename, 'w') 488 results_file.write(out)489 results_file.write(err)490 results_file.close()491 def run_noirotest_ml2_sanity(self):492 cmd = PPATH + 'cd noirotest/testcases/testcases_sanity && python run_ml2_sanity.py'493 out, err = self.remote_cmd(self.ssh_client, cmd, wait=ML2_SANITY_WAIT_TIME)494 self.save_results('ml2_sanity', out, err)495 def run_noirotest_gbp_sanity(self):496 cmd = PPATH + 'cd noirotest/testcases/testcases_sanity && python run_gbp_sanity.py'497 out, err = self.remote_cmd(self.ssh_client, cmd, wait=GBP_SANITY_WAIT_TIME)498 self.save_results('gbp_sanity', out, err)499 def run_noirotest_east_west_datapath(self):500 cmd = PPATH + 'cd noirotest/testcases/testcases_dp && python test_dp_runner.py'501 out, err = self.remote_cmd(self.ssh_client, cmd, wait=EAST_WEST_WAIT_TIME)502 self.save_results('east_west', out, err)503 def run_noirotest_north_south_datapath(self):504 cmd = PPATH + 'cd noirotest/testcases/testcases_nat_func && python test_gbp_nat_suite.py'505 out, err = self.remote_cmd(self.ssh_client, cmd, wait=NORTH_SOUTH_WAIT_TIME)506 self.save_results('north_south', out, err)507 def run_noirotest(self):508 self.run_noirotest_ml2_sanity()509 self.run_noirotest_gbp_sanity()510 self.run_noirotest_east_west_datapath()511 self.run_noirotest_north_south_datapath()512 def cleanup(self, host):513 pass514 #ssh_client = self.get_ssh_client(host)515 #fipsubnet1 = '50.50.50.0/24'516 #fipsubnet2 = '60.60.60.0/24'517 #sudo("ip route del %s " %(route))518 #sudo("ip route add %s via %s" %(route,nexthop))519class NautoPostDeployRunner(Runner):520 """Class to fix issues with Nauto deployments.521 The nauto deployer has been deprecated in favor of OpenStack522 Director (OSD) deploys. As a result, certain bugs have crept in to523 the nauto deployer since it is no longer supported. However,524 until the OSD deployer is ready, we will still use nauto for this,525 and therefore need to fix the issues from a given deploy.526 """527 def __init__(self, controller = None,528 compute1 = None, compute2 = None, ext_rtr = None):529 self.controller_host = controller530 self.compute1_host = compute1531 self.compute2_host = compute2532 self.external_router = ext_rtr533 self.hosts = [self.external_router, self.controller_host,534 self.compute1_host, self.compute2_host]535 self.image_uuid = None536 self.network_uuid = None537 super(NautoPostDeployRunner, self).__init__()538 def cleanup_dead_agents(self):539 ssh_client = self.get_ssh_client(self.controller_host)540 cmd = self.KEY + "for agent in `neutron agent-list | grep xxx | awk -F'|' '{print $2}'`; do neutron agent-delete $agent; done"541 print cmd542 self.remote_cmd(ssh_client, cmd)543 def get_dhcp_hosts(self):544 """Return compute hosts running DHCP agent.545 546 Return all compute hosts running the DHCP agent.547 """548 ssh_client = self.get_ssh_client(self.controller_host)549 cmd = self.KEY + "neutron agent-list | grep dhcp | awk -F'|' '{print $4}'"550 print cmd551 hosts, _ = self.remote_cmd(ssh_client, cmd)552 return [host.strip() for host in hosts.split("\n") if host]553 def update_dhcp_agent_cfg(self, ssh_client):554 """Fix the neutron-dhcp-agent configuration..555 556 Our nauto installer has a bug where it fails to configure557 the DHCP agent to use force_metadata (needed because there is558 no reference implementation neutron router in the network).559 This sets the force_metadata option to True and restarts560 the neutron-dhcp-agent service.561 """562 SED_FIND = '#force_metadata = false'563 SED_REPLACE = 'force_metadata = true'564 FILE = '/etc/neutron/dhcp_agent.ini'565 cmd = "sed -i 's/" + SED_FIND + "/" + SED_REPLACE + "/g' " + FILE566 print cmd567 self.remote_cmd(ssh_client, cmd)568 cmd = "service neutron-dhcp-agent restart"569 print cmd570 self.remote_cmd(ssh_client, cmd)571 def restart_metadata_agent(self, ssh_client):572 """Restart metdata agent.573 574 Our nauto installer has a bug where it fails to configure575 the metadata agent correctly on all compute hosts except576 the first. This function kills supervisord on the host,577 deletes all the flows on br-fabric, and restarts the neutron-opflex-agent,578 which will use create new metadata agents using the updated579 configuration file.580 """581 cmd = "kill `ps -ef | grep [s]upervisord | awk -F' ' '{print $2}'`"582 print cmd583 self.remote_cmd(ssh_client, cmd)584 cmd = "ovs-ofctl del-flows br-fabric"585 print cmd586 self.remote_cmd(ssh_client, cmd)587 cmd = "service neutron-opflex-agent restart"588 print cmd589 self.remote_cmd(ssh_client, cmd)590 def fix_br_fabric(self, ssh_client):591 cmd1 = "ovs-vsctl --may-exist add-port br-fabric br-fab_vxlan0 -- "592 cmd2 = "set Interface br-fab_vxlan0 type=vxlan options:remote_ip=flow "593 cmd3 = "options:key=flow options:dst_port=8472"594 cmd = cmd1 + cmd2 + cmd3595 print cmd596 self.remote_cmd(ssh_client, cmd)597 def restart_agent_ovs(self, ssh_client):598 cmd = "service agent-ovs restart"599 print cmd600 self.remote_cmd(ssh_client, cmd)601 def clear_iptables_rule(self, ssh_client):602 cmd = CLEAR_IP_TABLES_RULE603 print cmd604 self.remote_cmd(ssh_client, cmd)605 def update_agent_ovs_cfg(self, ssh_client):606 """Fix the agent-ovs configuration..607 608 Our nauto installer has a bug where it configures609 the SSL option as "enabled" in agent-ovs, which is610 an invalid value. This function replaces enabled611 with the correct value of "encrypted"612 """613 SED_FIND = '"mode": "enabled"'614 SED_REPLACE = '"mode": "encrypted"'615 FILE = '/etc/opflex-agent-ovs/conf.d/opflex-agent-ovs.conf'616 cmd = "sed -i 's/" + SED_FIND + "/" + SED_REPLACE + "/g' " + FILE617 print cmd618 self.remote_cmd(ssh_client, cmd)619 def get_compute_hosts(self):620 """Return all the compute hosts in the deployment. """621 ssh_client = self.get_ssh_client(self.controller_host)622 cmd = self.KEY + "nova service-list | grep nova-compute | awk -F'|' '{print $4}'"623 print cmd624 hosts, _ = self.remote_cmd(ssh_client, cmd)625 return [host.strip() for host in hosts.split("\n") if host]626 def update_metadata_agent_cfg(self, hosts):627 """Fix the neutron-metaata-agent configuration.628 629 Our nauto installer has a bug where it fails to configure630 the metadata agent correctly on all compute hosts except631 the first. This scp's the configuration file from the first632 compute host to all the others633 """634 ssh_client = self.get_ssh_client(self.controller_host)635 # get the file from the first compute host636 PATH = '/etc/neutron/'637 FILE = 'metadata_agent.ini'638 cmd = ('scp root@%(host)s:%(path)s%(file)s .' %639 {'host': hosts[0], 'file': FILE, 'path': PATH})640 print cmd641 self.remote_cmd(ssh_client, cmd)642 # now copy it back down to the other compute hosts643 for host in hosts[1:]:644 cmd = ("scp %(file)s root@%(host)s:%(path)s%(file)s" %645 {'host': host, 'file': FILE, 'path': PATH})646 print cmd647 self.remote_cmd(ssh_client, cmd)648 def apply_fixes(self):649 self.setup_passwordless_ssh(user='noiro', hosts=[self.external_router])650 self.setup_passwordless_ssh(hosts=[self.controller_host,651 self.compute1_host,652 self.compute2_host])653 self.cleanup_dead_agents()654 dhcp_host = self.get_dhcp_hosts()[0]655 print "DHCP host is " + dhcp_host656 all_hosts = self.get_compute_hosts()657 self.update_metadata_agent_cfg(all_hosts)658 for host in all_hosts:659 ssh_client = self.get_ssh_client(host)660 if host == dhcp_host:661 self.update_dhcp_agent_cfg(ssh_client)662 self.update_agent_ovs_cfg(ssh_client)663 if host != all_hosts[0]:664 self.restart_metadata_agent(ssh_client)665 self.fix_br_fabric(ssh_client)666 self.restart_agent_ovs(ssh_client)667 self.clear_iptables_rule(ssh_client)668@click.command()669@click.option('--controller-ip',670 help='IP address of OpenStack controller')671@click.option('--router-ip',672 help='IP address of External Router VM')673@click.option('--undercloud-type', default='director',674 help='Type of undercloud (juju or director)')675def make_tempest_config(controller_ip, router_ip, undercloud_type):676 if not controller_ip:677 try:678 fd = open('./controller_ip.txt', 'r')679 controller_ip = fd.readline().strip()680 except IOError as e:681 print("Couldn't open %s" % './controller_ip.txt')682 sys.exit(0)683 if not router_ip:684 try:685 fd = open('./router_ip.txt', 'r')686 router_ip = fd.readline().strip()687 except IOError as e:688 print("Couldn't open %s" % './router_ip.txt')689 sys.exit(0)690 runner = TempestTestRunner(controller=controller_ip, ext_rtr=router_ip, undercloud_type=undercloud_type)691 runner.configure_setup_for_test()692 click.echo("Iniitialization complete, and tempest config generated")693if __name__ == '__main__':...
exclude_tests.py
Source:exclude_tests.py
1#!/usr/bin/python2import click3import subprocess4JUJU='juju'5JUJU_RC='admin-openrc.sh'6DIRECTOR='director'7DIRECTOR_RC='overcloudrc'8# Dictionary that shows which tempest tests to exclude, by release9excluded_test_matrix = {10 'stable/newton': [11 ('tempest.api.network.test_floating_ips.FloatingIPTestJSON.test_floating_ip_update_different_router',12 'Test uses overlapping CIDRs in the same VRF'),13 ('tempest.api.network.test_networks.NetworksTest.test_external_network_visibility',14 'BUG: https://github.com/noironetworks/support/issues/916'),15 ('tempest.api.network.test_security_groups.SecGroupTest.test_create_show_delete_security_group_rule',16 'BUG: https://github.com/noironetworks/support/issues/710'),17 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_subnet_details',18 "BIG: https://github.com/noironetworks/support/issues/915"),19 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_update_router_admin_state',20 'BUG: https://github.com/noironetworks/support/issues/491'),21 ('neutron.tests.tempest.api.test_floating_ips_negative.FloatingIPNegativeTestJSON.test_associate_floatingip_with_port_with_floatingip',22 'Test uses overlapping CIDRs in the same VRF'),23 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sg_with_timestamp',24 'BUG: https://github.com/noironetworks/support/issues/710'),25 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sgrule_with_timestamp',26 'BUG: https://github.com/noironetworks/support/issues/710'),27 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sg_attribute_with_timestamp',28 'BUG: https://github.com/noironetworks/support/issues/710'),29 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sgrule_attribute_with_timestamp',30 'BUG: https://github.com/noironetworks/support/issues/710'),31 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_update_sg_with_timestamp',32 'BUG: https://github.com/noironetworks/support/issues/710')33 ],34 'stable/ocata': [35 ('tempest.api.network.test_floating_ips.FloatingIPTestJSON.test_floating_ip_update_different_router',36 'Test uses overlapping CIDRs in the same VRF'),37 ('tempest.api.network.test_networks.NetworksTest.test_external_network_visibility',38 'BUG: https://github.com/noironetworks/support/issues/916'),39 ('tempest.api.network.test_security_groups.SecGroupTest.test_create_show_delete_security_group_rule',40 'BUG: https://github.com/noironetworks/support/issues/710'),41 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_subnet_details',42 "BIG: https://github.com/noironetworks/support/issues/915"),43 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_update_router_admin_state',44 'BUG: https://github.com/noironetworks/support/issues/491'),45 ('neutron.tests.tempest.api.test_floating_ips_negative.FloatingIPNegativeTestJSON.test_associate_floatingip_with_port_with_floatingip',46 'Test uses overlapping CIDRs in the same VRF'),47 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sg_with_timestamp',48 'BUG: https://github.com/noironetworks/support/issues/710'),49 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sgrule_with_timestamp',50 'BUG: https://github.com/noironetworks/support/issues/710'),51 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sg_attribute_with_timestamp',52 'BUG: https://github.com/noironetworks/support/issues/710'),53 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sgrule_attribute_with_timestamp',54 'BUG: https://github.com/noironetworks/support/issues/710'),55 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_update_sg_with_timestamp',56 'BUG: https://github.com/noironetworks/support/issues/710')57 ],58 'stable/pike': [59 ('tempest.api.network.test_floating_ips.FloatingIPTestJSON.test_floating_ip_update_different_router',60 'Test uses overlapping CIDRs in the same VRF'),61 ('tempest.api.network.test_networks.NetworksTest.test_external_network_visibility',62 'BUG: https://github.com/noironetworks/support/issues/916'),63 ('tempest.api.network.test_security_groups.SecGroupTest.test_create_show_delete_security_group_rule',64 'BUG: https://github.com/noironetworks/support/issues/710'),65 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_subnet_details',66 "BIG: https://github.com/noironetworks/support/issues/915"),67 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_update_router_admin_state',68 'BUG: https://github.com/noironetworks/support/issues/491'),69 ('neutron.tests.tempest.api.test_floating_ips_negative.FloatingIPNegativeTestJSON.test_associate_floatingip_with_port_with_floatingip',70 'Test uses overlapping CIDRs in the same VRF'),71 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sg_with_timestamp',72 'BUG: https://github.com/noironetworks/support/issues/710'),73 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_create_sgrule_with_timestamp',74 'BUG: https://github.com/noironetworks/support/issues/710'),75 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sg_attribute_with_timestamp',76 'BUG: https://github.com/noironetworks/support/issues/710'),77 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_show_sgrule_attribute_with_timestamp',78 'BUG: https://github.com/noironetworks/support/issues/710'),79 ('neutron.tests.tempest.api.test_timestamp.TestTimeStampWithSecurityGroup.test_update_sg_with_timestamp',80 'BUG: https://github.com/noironetworks/support/issues/710')81 ],82 'stable/queens': [83 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_subnet_details',84 "BIG: https://github.com/noironetworks/support/issues/915"),85 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_update_router_admin_state',86 'BUG: https://github.com/noironetworks/support/issues/491'),87 ('tempest.scenario.test_network_basic_ops.TestNetworkBasicOps.test_port_security_macspoofing_port',88 'Why is this failing?'),89 ('tempest.api.network.test_floating_ips.FloatingIPTestJSON.test_floating_ip_update_different_router',90 'Test uses overlapping CIDRs in the same VRF'),91 ('tempest.api.compute.servers.test_attach_interfaces.AttachInterfacesTestJSON.test_create_list_show_delete_interfaces_by_fixed_ip',92 'Upstream bug: https://bugs.launchpad.net/tempest/+bug/1790864'),93 ('neutron_tempest_plugin.api.test_revisions.TestRevisions.test_update_network_constrained_by_revision',94 'Why is this test failing?'),95 ('neutron_tempest_plugin.api.test_timestamp.TestTimeStampWithL3.test_show_floatingip_attribute_with_timestamp',96 'Why is this test failing?'),97 ('neutron_tempest_plugin.scenario.test_connectivity.NetworkConnectivityTest.test_connectivity_through_2_routers',98 'The apic_aim mechanism driver does not support transit routes (connect networks to same neutron router instead)'),99 ('neutron_tempest_plugin.scenario.test_internal_dns.InternalDNSTest.test_dns_domain_and_name',100 'This is an upstream bug (non-optimized has same failure)')101 ],102}103class Excluder(object):104 def __init__(self, undercloud_type=DIRECTOR):105 if undercloud_type == DIRECTOR:106 self.KEY = 'source ~/' + DIRECTOR_RC + ' && '107 elif undercloud_type == JUJU:108 self.KEY = 'source ~/' + JUJU_RC + ' && '109 self.KEY = 'source ~/' + DIRECTOR_RC + ' && '110 self.version = self.get_openstack_version()111 def get_openstack_version(self):112 # we rely on the openstack client repo113 cmd1 = "cd python-openstackclient/ && "114 cmd2 = "git status | grep 'branch stable\|eol' "115 cmd3 = "| awk 'NF>1{print $NF}' && cd .."116 cmd = self.KEY + cmd1 + cmd2 + cmd3117 print cmd118 subprocess.check_output(['bash','-c', cmd])119 version_string = subprocess.check_output(['bash','-c', cmd])120 version_list = [version.strip()121 for version in version_string.split("\n") if version]122 version = version_list[0]123 if version == 'newton-eol':124 version = 'stable/newton'125 return version126 127 def get_test_info(self, test):128 test_pieces = test.split(".")129 path = "-".join(test_pieces[0].split("_")) + "/" + "/".join(test_pieces[0:-2]) + ".py"130 test_class = test_pieces[-2:-1][0]131 test_name = test.split(".")[-1:][0]132 return path, test_class, test_name133 def exclude_tests(self):134 for test, reason in excluded_test_matrix[self.version]:135 path, test_class, test_name = self.get_test_info(test)136 def add_line(file_name, before_line, new_line):137 # find the first instance of the string138 index = None139 all_lines = None140 with open(file_name, "r") as fd:141 all_lines = fd.readlines()142 for count in range(len(all_lines)):143 line = all_lines[count]144 if before_line in line:145 strip_line = line[:-1]146 indent_len = len(strip_line) - len(strip_line.strip())147 indent = line[:indent_len] if indent_len else ''148 new_line = indent + new_line149 index = count150 break151 if index:152 all_lines.insert(index, new_line)153 with open(file_name, "w+") as fd:154 contents = "".join(all_lines)155 fd.write(contents)156 157 cmd = "egrep testtools %s" % path158 try:159 subprocess.check_output(['bash','-c', cmd])160 except Exception as e:161 # not present, so add testtools162 add_line(path, "import", "import testtools\n")163 pass164 # need to escape backslashes165 166 reason_str = '@testtools.skip("' + reason + '")\n'167 # Skip this test168 add_line(path, 'def ' + test_name, reason_str)169@click.command()170@click.option('--undercloud-type', default='director',171 help='Type of undercloud (juju or director)')172def exclude_tests(undercloud_type):173 excluder = Excluder(undercloud_type=undercloud_type)174 excluder.exclude_tests()175 click.echo("Iniitialization complete, and tempest config generated")176if __name__ == '__main__':...
test_network_basic_ops.py
Source:test_network_basic_ops.py
1# Copyright 2016 VMware Inc.2# All Rights Reserved.3#4# Licensed under the Apache License, Version 2.0 (the "License"); you may5# not use this file except in compliance with the License. You may obtain6# a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13# License for the specific language governing permissions and limitations14# under the License.15from tempest.scenario import test_network_basic_ops as network_ops16class TestNetworkBasicOps(network_ops.TestNetworkBasicOps):17 # NSX-v does not allow tenants to access dhcp service.18 # Overwirte parent class to skip dhcp service testing.19 def _list_ports(self, *args, **kwargs):20 """List ports using admin creds """21 ports_list = self.admin_manager.ports_client.list_ports(22 *args, **kwargs)23 return ports_list['ports']24 def _check_network_internal_connectivity(self, network,25 should_connect=True):26 floating_ip, server = self.floating_ip_tuple27 # get internal ports' ips:28 # get all network ports in the new network29 # NSX-v: dhcp is not reachable30 internal_ips = (p['fixed_ips'][0]['ip_address'] for p in31 self._list_ports(tenant_id=server['tenant_id'],32 network_id=network['id'])33 if (p['device_owner'].startswith('network') and34 not p['device_owner'].endswith('dhcp')))35 self._check_server_connectivity(floating_ip,36 internal_ips,...
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 tempest 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?
