diff options
| -rw-r--r-- | functest/opnfv_tests/openstack/vping/vping_base.py | 183 | ||||
| -rw-r--r-- | functest/opnfv_tests/openstack/vping/vping_ssh.py | 194 | ||||
| -rw-r--r-- | functest/opnfv_tests/openstack/vping/vping_userdata.py | 128 | ||||
| -rw-r--r-- | functest/tests/unit/openstack/vping/test_vping.py | 179 | ||||
| -rw-r--r-- | requirements.txt | 2 |
5 files changed, 204 insertions, 482 deletions
diff --git a/functest/opnfv_tests/openstack/vping/vping_base.py b/functest/opnfv_tests/openstack/vping/vping_base.py index 4b6e4e97f..246b12c97 100644 --- a/functest/opnfv_tests/openstack/vping/vping_base.py +++ b/functest/opnfv_tests/openstack/vping/vping_base.py @@ -15,18 +15,10 @@ import logging import time import uuid -from snaps.config.flavor import FlavorConfig -from snaps.config.network import NetworkConfig -from snaps.config.network import SubnetConfig -from snaps.config.router import RouterConfig -from snaps.openstack.create_flavor import OpenStackFlavor -from snaps.openstack.create_image import OpenStackImage -from snaps.openstack.create_network import OpenStackNetwork -from snaps.openstack.create_router import OpenStackRouter -from snaps.openstack.tests import openstack_tests +import os_client_config +import shade from xtesting.core import testcase -from functest.opnfv_tests.openstack.snaps import snaps_utils from functest.utils import config from functest.utils import env @@ -43,42 +35,24 @@ class VPingBase(testcase.TestCase): def __init__(self, **kwargs): super(VPingBase, self).__init__(**kwargs) self.logger = logging.getLogger(__name__) - self.os_creds = kwargs.get('os_creds') or snaps_utils.get_credentials() - self.creators = list() - self.image_creator = None - self.network_creator = None - self.vm1_creator = None - self.vm2_creator = None - self.router_creator = None - - # Shared metadata + self.cloud = shade.OperatorCloud( + cloud_config=os_client_config.get_config()) + self.ext_net = self.cloud.get_network("ext-net") + self.logger.debug("ext_net: %s", self.ext_net) self.guid = '-' + str(uuid.uuid4()) - - self.router_name = getattr( - config.CONF, 'vping_router_name') + self.guid - self.vm1_name = getattr( - config.CONF, 'vping_vm_name_1') + self.guid - self.vm2_name = getattr(config.CONF, 'vping_vm_name_2') + self.guid - - self.vm_boot_timeout = getattr(config.CONF, 'vping_vm_boot_timeout') - self.vm_delete_timeout = getattr( - config.CONF, 'vping_vm_delete_timeout') - self.vm_ssh_connect_timeout = getattr( - config.CONF, 'vping_vm_ssh_connect_timeout') - self.ping_timeout = getattr(config.CONF, 'vping_ping_timeout') - self.flavor_name = 'vping-flavor' + self.guid - - # Move this configuration option up for all tests to leverage - if hasattr(config.CONF, 'snaps_images_cirros'): - self.cirros_image_config = getattr( - config.CONF, 'snaps_images_cirros') - else: - self.cirros_image_config = None + self.network = None + self.subnet = None + self.router = None + self.image = None + self.flavor = None + self.vm1 = None def run(self, **kwargs): # pylint: disable=too-many-locals """ Begins the test execution which should originate from the subclass """ + assert self.cloud + assert self.ext_net self.logger.info('Begin virtual environment setup') self.start_time = time.time() @@ -89,14 +63,11 @@ class VPingBase(testcase.TestCase): image_base_name = '{}-{}'.format( getattr(config.CONF, 'vping_image_name'), self.guid) - os_image_settings = openstack_tests.cirros_image_settings( - image_base_name, image_metadata=self.cirros_image_config) self.logger.info("Creating image with name: '%s'", image_base_name) - - self.image_creator = OpenStackImage( - self.os_creds, os_image_settings) - self.image_creator.create() - self.creators.append(self.image_creator) + self.image = self.cloud.create_image( + image_base_name, + filename=getattr(config.CONF, 'openstack_image_url')) + self.logger.debug("image: %s", self.image) private_net_name = getattr( config.CONF, 'vping_private_net_name') + self.guid @@ -104,59 +75,61 @@ class VPingBase(testcase.TestCase): config.CONF, 'vping_private_subnet_name') + self.guid) private_subnet_cidr = getattr(config.CONF, 'vping_private_subnet_cidr') - vping_network_type = None - vping_physical_network = None - vping_segmentation_id = None - + provider = {} if hasattr(config.CONF, 'vping_network_type'): - vping_network_type = getattr(config.CONF, 'vping_network_type') + provider["network_type"] = getattr( + config.CONF, 'vping_network_type') if hasattr(config.CONF, 'vping_physical_network'): - vping_physical_network = getattr( + provider["physical_network"] = getattr( config.CONF, 'vping_physical_network') if hasattr(config.CONF, 'vping_segmentation_id'): - vping_segmentation_id = getattr( + provider["segmentation_id"] = getattr( config.CONF, 'vping_segmentation_id') - self.logger.info( "Creating network with name: '%s'", private_net_name) - subnet_settings = SubnetConfig( - name=private_subnet_name, + self.network = self.cloud.create_network( + private_net_name, + provider=provider) + self.logger.debug("network: %s", self.network) + + self.subnet = self.cloud.create_subnet( + self.network.id, + subnet_name=private_subnet_name, cidr=private_subnet_cidr, + enable_dhcp=True, dns_nameservers=[env.get('NAMESERVER')]) - self.network_creator = OpenStackNetwork( - self.os_creds, - NetworkConfig( - name=private_net_name, - network_type=vping_network_type, - physical_network=vping_physical_network, - segmentation_id=vping_segmentation_id, - subnet_settings=[subnet_settings])) - self.network_creator.create() - self.creators.append(self.network_creator) - - # Creating router to external network - self.logger.info("Creating router with name: '%s'", self.router_name) - ext_net_name = snaps_utils.get_ext_net_name(self.os_creds) - self.router_creator = OpenStackRouter( - self.os_creds, - RouterConfig( - name=self.router_name, - external_gateway=ext_net_name, - internal_subnets=[subnet_settings.name])) - self.router_creator.create() - self.creators.append(self.router_creator) + self.logger.debug("subnet: %s", self.subnet) - self.logger.info( - "Creating flavor with name: '%s'", self.flavor_name) - flavor_ram = getattr(config.CONF, 'openstack_flavor_ram') - flavor_metadata = getattr(config.CONF, 'flavor_extra_specs', None) + router_name = getattr(config.CONF, 'vping_router_name') + self.guid + self.logger.info("Creating router with name: '%s'", router_name) + self.router = self.cloud.create_router( + name=router_name, + ext_gateway_net_id=self.ext_net.id) + self.logger.debug("router: %s", self.router) + self.cloud.add_router_interface(self.router, subnet_id=self.subnet.id) - flavor_creator = OpenStackFlavor( - self.os_creds, - FlavorConfig(name=self.flavor_name, ram=flavor_ram, disk=1, - vcpus=1, metadata=flavor_metadata)) - flavor_creator.create() - self.creators.append(flavor_creator) + flavor_name = 'vping-flavor' + self.guid + self.logger.info( + "Creating flavor with name: '%s'", flavor_name) + self.flavor = self.cloud.create_flavor( + flavor_name, getattr(config.CONF, 'openstack_flavor_ram'), + getattr(config.CONF, 'openstack_flavor_vcpus'), + getattr(config.CONF, 'openstack_flavor_disk')) + self.logger.debug("flavor: %s", self.flavor) + self.cloud.set_flavor_specs( + self.flavor.id, getattr(config.CONF, 'flavor_extra_specs', {})) + + vm1_name = getattr(config.CONF, 'vping_vm_name_1') + self.guid + self.logger.info( + "Creating VM 1 instance with name: '%s'", vm1_name) + self.vm1 = self.cloud.create_server( + vm1_name, image=self.image.id, + flavor=self.flavor.id, + auto_ip=False, wait=True, + timeout=getattr(config.CONF, 'vping_vm_boot_timeout'), + network=self.network.id) + self.logger.debug("vm1: %s", self.vm1) + self.vm1 = self.cloud.wait_for_server(self.vm1, auto_ip=False) def _execute(self): """ @@ -164,22 +137,11 @@ class VPingBase(testcase.TestCase): :return: the exit code """ self.logger.info('Begin test execution') - - test_ip = self.vm1_creator.get_port_ip( - self.vm1_creator.instance_settings.port_settings[0].name) - - if self.vm1_creator.vm_active( - block=True) and self.vm2_creator.vm_active(block=True): - result = self._do_vping(self.vm2_creator, test_ip) - else: - raise Exception('VMs never became active') - + result = self._do_vping() self.stop_time = time.time() - if result != testcase.TestCase.EX_OK: self.result = 0 return testcase.TestCase.EX_RUN_ERROR - self.result = 100 return testcase.TestCase.EX_OK @@ -188,19 +150,18 @@ class VPingBase(testcase.TestCase): Cleanup all OpenStack objects. Should be called on completion :return: """ - if getattr(config.CONF, 'vping_cleanup_objects') == 'True': - for creator in reversed(self.creators): - try: - creator.clean() - except Exception: # pylint: disable=broad-except - self.logger.exception('Unexpected error cleaning') - - def _do_vping(self, vm_creator, test_ip): + assert self.cloud + self.cloud.delete_server(self.vm1, wait=True) + self.cloud.delete_image(self.image) + self.cloud.remove_router_interface(self.router, self.subnet.id) + self.cloud.delete_router(self.router.id) + self.cloud.delete_network(self.network.id) + self.cloud.delete_flavor(self.flavor.id) + + def _do_vping(self): """ Method to be implemented by subclasses Begins the real test after the OpenStack environment has been setup - :param vm_creator: the SNAPS VM instance creator object - :param test_ip: the IP to which the VM needs to issue the ping :return: T/F """ raise NotImplementedError('vping execution is not implemented') diff --git a/functest/opnfv_tests/openstack/vping/vping_ssh.py b/functest/opnfv_tests/openstack/vping/vping_ssh.py index 93c9c0976..21a433884 100644 --- a/functest/opnfv_tests/openstack/vping/vping_ssh.py +++ b/functest/opnfv_tests/openstack/vping/vping_ssh.py @@ -10,18 +10,13 @@ """vPingSSH testcase.""" import logging +import os +import tempfile import time -from scp import SCPClient +import paramiko import pkg_resources - -from snaps.config.keypair import KeypairConfig -from snaps.config.network import PortConfig -from snaps.config.security_group import ( - Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig) -from snaps.config.vm_inst import FloatingIpConfig, VmInstanceConfig -from snaps.openstack.utils import deploy_utils - +from scp import SCPClient from xtesting.core import testcase from xtesting.energy import energy @@ -30,7 +25,6 @@ from functest.utils import config class VPingSSH(vping_base.VPingBase): - # pylint: disable=too-many-instance-attributes """ VPingSSH testcase implementation. @@ -44,12 +38,12 @@ class VPingSSH(vping_base.VPingBase): kwargs["case_name"] = "vping_ssh" super(VPingSSH, self).__init__(**kwargs) self.logger = logging.getLogger(__name__) - - self.kp_name = getattr(config.CONF, 'vping_keypair_name') + self.guid - self.kp_priv_file = getattr(config.CONF, 'vping_keypair_priv_file') - self.kp_pub_file = getattr(config.CONF, 'vping_keypair_pub_file') - self.sg_name = getattr(config.CONF, 'vping_sg_name') + self.guid - self.sg_desc = getattr(config.CONF, 'vping_sg_desc') + self.vm2 = None + self.sec = None + self.fip = None + self.keypair = None + self.ssh = paramiko.SSHClient() + (_, self.key_filename) = tempfile.mkstemp() @energy.enable_recording def run(self, **kwargs): @@ -61,87 +55,67 @@ class VPingSSH(vping_base.VPingBase): :return: the exit code from the super.execute() method """ try: + assert self.cloud super(VPingSSH, self).run() - log = "Creating keypair with name: '%s'" % self.kp_name - self.logger.info(log) - kp_creator = deploy_utils.create_keypair( - self.os_creds, - KeypairConfig( - name=self.kp_name, private_filepath=self.kp_priv_file, - public_filepath=self.kp_pub_file)) - self.creators.append(kp_creator) - - # Creating Instance 1 - port1_settings = PortConfig( - name=self.vm1_name + '-vPingPort', - network_name=self.network_creator.network_settings.name) - instance1_settings = VmInstanceConfig( - name=self.vm1_name, flavor=self.flavor_name, - vm_boot_timeout=self.vm_boot_timeout, - vm_delete_timeout=self.vm_delete_timeout, - ssh_connect_timeout=self.vm_ssh_connect_timeout, - port_settings=[port1_settings]) - - log = ("Creating VM 1 instance with name: '%s'" - % instance1_settings.name) - self.logger.info(log) - self.vm1_creator = deploy_utils.create_vm_instance( - self.os_creds, - instance1_settings, - self.image_creator.image_settings, - keypair_creator=kp_creator) - self.creators.append(self.vm1_creator) - - # Creating Instance 2 - sg_creator = self.__create_security_group() - self.creators.append(sg_creator) - - port2_settings = PortConfig( - name=self.vm2_name + '-vPingPort', - network_name=self.network_creator.network_settings.name) - instance2_settings = VmInstanceConfig( - name=self.vm2_name, flavor=self.flavor_name, - vm_boot_timeout=self.vm_boot_timeout, - vm_delete_timeout=self.vm_delete_timeout, - ssh_connect_timeout=self.vm_ssh_connect_timeout, - port_settings=[port2_settings], - security_group_names=[sg_creator.sec_grp_settings.name], - floating_ip_settings=[FloatingIpConfig( - name=self.vm2_name + '-FIPName', - port_name=port2_settings.name, - router_name=self.router_creator.router_settings.name)]) - - log = ("Creating VM 2 instance with name: '%s'" - % instance2_settings.name) - self.logger.info(log) - self.vm2_creator = deploy_utils.create_vm_instance( - self.os_creds, - instance2_settings, - self.image_creator.image_settings, - keypair_creator=kp_creator) - self.creators.append(self.vm2_creator) + kp_name = getattr(config.CONF, 'vping_keypair_name') + self.guid + self.logger.info("Creating keypair with name: '%s'", kp_name) + self.keypair = self.cloud.create_keypair(kp_name) + self.logger.debug("keypair: %s", self.keypair) + self.logger.debug("private_key: %s", self.keypair.private_key) + with open(self.key_filename, 'w') as private_key_file: + private_key_file.write(self.keypair.private_key) + + self.sec = self.cloud.create_security_group( + getattr(config.CONF, 'vping_sg_name') + self.guid, + getattr(config.CONF, 'vping_sg_desc')) + self.cloud.create_security_group_rule( + self.sec.id, port_range_min='22', port_range_max='22', + protocol='tcp', direction='ingress') + self.cloud.create_security_group_rule( + self.sec.id, protocol='icmp', direction='ingress') + + vm2_name = "{}-{}-{}".format( + getattr(config.CONF, 'vping_vm_name_2'), "ssh", self.guid) + self.logger.info( + "Creating VM 2 instance with name: '%s'", vm2_name) + self.vm2 = self.cloud.create_server( + vm2_name, image=self.image.id, flavor=self.flavor.id, + key_name=self.keypair.id, + auto_ip=False, wait=True, + timeout=getattr(config.CONF, 'vping_vm_boot_timeout'), + network=self.network.id, + security_groups=[self.sec.id]) + self.logger.debug("vm2: %s", self.vm2) + self.fip = self.cloud.create_floating_ip( + network=self.ext_net.id, server=self.vm2) + self.logger.debug("floating_ip2: %s", self.fip) + self.vm2 = self.cloud.wait_for_server(self.vm2, auto_ip=False) return self._execute() except Exception: # pylint: disable=broad-except - self.logger.exception('Unexpected error running test') + self.logger.exception('Unexpected error running vping_ssh') return testcase.TestCase.EX_RUN_ERROR - def _do_vping(self, vm_creator, test_ip): + def _do_vping(self): """ Execute ping command. Override from super """ - if vm_creator.vm_ssh_active(block=True): - ssh = vm_creator.ssh_client() - if not self._transfer_ping_script(ssh): - return testcase.TestCase.EX_RUN_ERROR - return self._do_vping_ssh(ssh, test_ip) - else: + time.sleep(10) + self.ssh.set_missing_host_key_policy(paramiko.client.AutoAddPolicy()) + self.ssh.connect( + self.vm2.public_v4, + username=getattr(config.CONF, 'openstack_image_user'), + key_filename=self.key_filename, + timeout=getattr(config.CONF, 'vping_vm_ssh_connect_timeout')) + self.logger.debug("ssh: %s", self.ssh) + if not self._transfer_ping_script(): return testcase.TestCase.EX_RUN_ERROR + return self._do_vping_ssh() - def _transfer_ping_script(self, ssh): + def _transfer_ping_script(self): """ Transfert vping script to VM. @@ -150,7 +124,7 @@ class VPingSSH(vping_base.VPingBase): :return: """ self.logger.info("Trying to transfer ping.sh") - scp = SCPClient(ssh.get_transport()) + scp = SCPClient(self.ssh.get_transport()) ping_script = pkg_resources.resource_filename( 'functest.opnfv_tests.openstack.vping', 'ping.sh') try: @@ -160,14 +134,13 @@ class VPingSSH(vping_base.VPingBase): return False cmd = 'chmod 755 ~/ping.sh' - # pylint: disable=unused-variable - (stdin, stdout, stderr) = ssh.exec_command(cmd) + (_, stdout, _) = self.ssh.exec_command(cmd) for line in stdout.readlines(): print line return True - def _do_vping_ssh(self, ssh, test_ip): + def _do_vping_ssh(self): """ Execute ping command via SSH. @@ -180,12 +153,12 @@ class VPingSSH(vping_base.VPingBase): self.logger.info("Waiting for ping...") sec = 0 - cmd = '~/ping.sh ' + test_ip + cmd = '~/ping.sh ' + self.vm1.private_v4 flag = False while True: time.sleep(1) - (_, stdout, _) = ssh.exec_command(cmd) + (_, stdout, _) = self.ssh.exec_command(cmd) output = stdout.readlines() for line in output: @@ -194,43 +167,24 @@ class VPingSSH(vping_base.VPingBase): exit_code = testcase.TestCase.EX_OK flag = True break - - elif sec == self.ping_timeout: + elif sec == getattr(config.CONF, 'vping_ping_timeout'): self.logger.info("Timeout reached.") flag = True break if flag: break - log = "Pinging %s. Waiting for response..." % test_ip + log = "Pinging %s. Waiting for response..." % self.vm1.private_v4 self.logger.debug(log) sec += 1 return exit_code - def __create_security_group(self): - """ - Configure OpenStack security groups. - - Configures and deploys an OpenStack security group object - :return: the creator object - """ - sg_rules = list() - sg_rules.append( - SecurityGroupRuleConfig( - sec_grp_name=self.sg_name, direction=Direction.ingress, - protocol=Protocol.icmp)) - sg_rules.append( - SecurityGroupRuleConfig( - sec_grp_name=self.sg_name, direction=Direction.ingress, - protocol=Protocol.tcp, port_range_min=22, port_range_max=22)) - sg_rules.append( - SecurityGroupRuleConfig( - sec_grp_name=self.sg_name, direction=Direction.egress, - protocol=Protocol.tcp, port_range_min=22, port_range_max=22)) - - log = "Security group with name: '%s'" % self.sg_name - self.logger.info(log) - return deploy_utils.create_security_group(self.os_creds, - SecurityGroupConfig( - name=self.sg_name, - description=self.sg_desc, - rule_settings=sg_rules)) + def clean(self): + assert self.cloud + os.remove(self.key_filename) + self.cloud.delete_server( + self.vm2, wait=True, + timeout=getattr(config.CONF, 'vping_vm_delete_timeout')) + self.cloud.delete_security_group(self.sec.id) + self.cloud.delete_keypair(self.keypair.id) + self.cloud.delete_floating_ip(self.fip.id) + super(VPingSSH, self).clean() diff --git a/functest/opnfv_tests/openstack/vping/vping_userdata.py b/functest/opnfv_tests/openstack/vping/vping_userdata.py index c2ff75150..9551c4b5d 100644 --- a/functest/opnfv_tests/openstack/vping/vping_userdata.py +++ b/functest/opnfv_tests/openstack/vping/vping_userdata.py @@ -12,12 +12,10 @@ import logging import time -from snaps.config.network import PortConfig -from snaps.config.vm_inst import VmInstanceConfig -from snaps.openstack.utils import deploy_utils from xtesting.core import testcase from functest.opnfv_tests.openstack.vping import vping_base +from functest.utils import config class VPingUserdata(vping_base.VPingBase): @@ -30,6 +28,7 @@ class VPingUserdata(vping_base.VPingBase): kwargs["case_name"] = "vping_userdata" super(VPingUserdata, self).__init__(**kwargs) self.logger = logging.getLogger(__name__) + self.vm2 = None def run(self, **kwargs): """ @@ -37,53 +36,29 @@ class VPingUserdata(vping_base.VPingBase): validates. :return: the exit code from the super.execute() method """ - super(VPingUserdata, self).run() - - # Creating Instance 1 - port1_settings = PortConfig( - name=self.vm1_name + '-vPingPort', - network_name=self.network_creator.network_settings.name) - instance1_settings = VmInstanceConfig( - name=self.vm1_name, - flavor=self.flavor_name, - vm_boot_timeout=self.vm_boot_timeout, - port_settings=[port1_settings]) - - self.logger.info( - "Creating VM 1 instance with name: '%s'", - instance1_settings.name) - self.vm1_creator = deploy_utils.create_vm_instance( - self.os_creds, instance1_settings, - self.image_creator.image_settings) - self.creators.append(self.vm1_creator) - - userdata = _get_userdata( - self.vm1_creator.get_port_ip(port1_settings.name)) - if userdata: - # Creating Instance 2 - port2_settings = PortConfig( - name=self.vm2_name + '-vPingPort', - network_name=self.network_creator.network_settings.name) - instance2_settings = VmInstanceConfig( - name=self.vm2_name, - flavor=self.flavor_name, - vm_boot_timeout=self.vm_boot_timeout, - port_settings=[port2_settings], - userdata=userdata) + try: + assert self.cloud + super(VPingUserdata, self).run() + vm2_name = "{}-{}-{}".format( + getattr(config.CONF, 'vping_vm_name_2'), "userdata", self.guid) self.logger.info( - "Creating VM 2 instance with name: '%s'", - instance2_settings.name) - self.vm2_creator = deploy_utils.create_vm_instance( - self.os_creds, instance2_settings, - self.image_creator.image_settings) - self.creators.append(self.vm2_creator) - else: - raise Exception('Userdata is None') - - return self._execute() - - def _do_vping(self, vm_creator, test_ip): + "Creating VM 2 instance with name: '%s'", vm2_name) + self.vm2 = self.cloud.create_server( + vm2_name, image=self.image.id, flavor=self.flavor.id, + auto_ip=False, wait=True, + timeout=getattr(config.CONF, 'vping_vm_boot_timeout'), + network=self.network.id, + userdata=self._get_userdata()) + self.logger.debug("vm2: %s", self.vm2) + self.vm2 = self.cloud.wait_for_server(self.vm2, auto_ip=False) + + return self._execute() + except Exception: # pylint: disable=broad-except + self.logger.exception('Unexpected error running vping_userdata') + return testcase.TestCase.EX_RUN_ERROR + + def _do_vping(self): """ Override from super """ @@ -94,7 +69,8 @@ class VPingUserdata(vping_base.VPingBase): while True: time.sleep(1) - p_console = vm_creator.get_console_output() + p_console = self.cloud.get_server_console(self.vm2.id) + self.logger.debug("console: \n%s", p_console) if "vPing OK" in p_console: self.logger.info("vPing detected!") exit_code = testcase.TestCase.EX_OK @@ -102,7 +78,7 @@ class VPingUserdata(vping_base.VPingBase): elif "failed to read iid from metadata" in p_console or tries > 5: self.logger.info("Failed to read iid from metadata") break - elif sec == self.ping_timeout: + elif sec == getattr(config.CONF, 'vping_ping_timeout'): self.logger.info("Timeout reached.") break elif sec % 10 == 0: @@ -113,29 +89,37 @@ class VPingUserdata(vping_base.VPingBase): tries += 1 else: self.logger.debug( - "Pinging %s. Waiting for response...", test_ip) + "Pinging %s. Waiting for response...", + self.vm1.private_v4) sec += 1 return exit_code - -def _get_userdata(test_ip): - """ - Returns the post VM creation script to be added into the VM's userdata - :param test_ip: the IP value to substitute into the script - :return: the bash script contents - """ - if test_ip: - return ("#!/bin/sh\n\n" - "while true; do\n" - " ping -c 1 %s 2>&1 >/dev/null\n" - " RES=$?\n" - " if [ \"Z$RES\" = \"Z0\" ] ; then\n" - " echo 'vPing OK'\n" - " break\n" - " else\n" - " echo 'vPing KO'\n" - " fi\n" - " sleep 1\n" - "done\n" % str(test_ip)) - return None + def _get_userdata(self): + """ + Returns the post VM creation script to be added into the VM's userdata + :param test_ip: the IP value to substitute into the script + :return: the bash script contents + """ + if self.vm1.private_v4: + return ("#!/bin/sh\n\n" + "while true; do\n" + " ping -c 1 %s 2>&1 >/dev/null\n" + " RES=$?\n" + " if [ \"Z$RES\" = \"Z0\" ] ; then\n" + " echo 'vPing OK'\n" + " break\n" + " else\n" + " echo 'vPing KO'\n" + " fi\n" + " sleep 1\n" + "done\n" % str(self.vm1.private_v4)) + return None + + def clean(self): + assert self.cloud + self.cloud.delete_server( + self.vm2, wait=True, + timeout=getattr(config.CONF, 'vping_vm_delete_timeout')) + self.cloud.delete_server(self.vm2, wait=True) + super(VPingUserdata, self).clean() diff --git a/functest/tests/unit/openstack/vping/test_vping.py b/functest/tests/unit/openstack/vping/test_vping.py deleted file mode 100644 index adc0f69a6..000000000 --- a/functest/tests/unit/openstack/vping/test_vping.py +++ /dev/null @@ -1,179 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2017 Cable Television Laboratories, Inc. and others. -# -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# -# http://www.apache.org/licenses/LICENSE-2.0 - -# pylint: disable=missing-docstring - -import logging -import unittest - -import mock -from snaps.config.keypair import KeypairConfig -from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig -from snaps.config.router import RouterConfig -from snaps.config.security_group import SecurityGroupConfig -from snaps.config.vm_inst import VmInstanceConfig -from snaps.openstack.create_image import OpenStackImage -from snaps.openstack.create_instance import OpenStackVmInstance -from snaps.openstack.create_keypairs import OpenStackKeypair -from snaps.openstack.create_network import OpenStackNetwork -from snaps.openstack.create_router import OpenStackRouter -from snaps.openstack.create_security_group import OpenStackSecurityGroup -from snaps.openstack.os_credentials import OSCreds -from xtesting.core import testcase - -from functest.opnfv_tests.openstack.vping import vping_userdata, vping_ssh -from functest.utils import env - - -class VPingUserdataTesting(unittest.TestCase): - """ - Ensures the VPingUserdata class can run in Functest. This test does not - actually connect with an OpenStack pod. - """ - - def setUp(self): - self.os_creds = OSCreds( - username='user', password='pass', - auth_url='http://foo.com:5000/v3', project_name='bar') - - self.vping_userdata = vping_userdata.VPingUserdata( - os_creds=self.os_creds) - - @mock.patch('snaps.openstack.utils.deploy_utils.create_vm_instance') - @mock.patch('os.path.exists', return_value=True) - @mock.patch('snaps.openstack.utils.keystone_utils.keystone_client') - @mock.patch('snaps.openstack.utils.keystone_utils.get_project') - @mock.patch('snaps.openstack.create_flavor.OpenStackFlavor.create', - return_value=None) - @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' - 'get_port_ip', return_value='10.0.0.1') - @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' - 'vm_active', return_value=True) - @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.' - 'get_ext_net_name', return_value='foo') - def test_vping_userdata(self, *args): - # pylint: disable=unused-argument - with mock.patch('snaps.openstack.create_image.OpenStackImage.create', - return_value=OpenStackImage(self.os_creds, None)), \ - mock.patch( - 'snaps.openstack.create_network.OpenStackNetwork.create', - return_value=OpenStackNetwork( - self.os_creds, NetworkConfig(name='foo'))), \ - mock.patch( - 'snaps.openstack.create_router.OpenStackRouter.create', - return_value=OpenStackRouter( - self.os_creds, RouterConfig(name='foo'))), \ - mock.patch('snaps.openstack.utils.deploy_utils.' - 'create_vm_instance', - return_value=OpenStackVmInstance( - self.os_creds, - VmInstanceConfig( - name='foo', flavor='bar', - port_settings=[PortConfig( - name='foo', network_name='bar')]), - None)), \ - mock.patch('snaps.openstack.create_instance.' - 'OpenStackVmInstance.get_console_output', - return_value='vPing OK'): - self.assertEquals( - testcase.TestCase.EX_OK, self.vping_userdata.run()) - - -class VPingSSHTesting(unittest.TestCase): - """ - Ensures the VPingUserdata class can run in Functest. This test does not - actually connect with an OpenStack pod. - """ - - def setUp(self): - self.os_creds = OSCreds( - username='user', password='pass', - auth_url='http://foo.com:5000/v3', project_name='bar') - - self.vping_ssh = vping_ssh.VPingSSH( - os_creds=self.os_creds) - - @mock.patch('snaps.openstack.utils.deploy_utils.create_vm_instance') - @mock.patch('os.path.exists', return_value=True) - @mock.patch('snaps.openstack.utils.keystone_utils.keystone_client') - @mock.patch('snaps.openstack.utils.keystone_utils.get_project') - @mock.patch('snaps.openstack.create_flavor.OpenStackFlavor.create', - return_value=None) - @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' - 'get_port_ip', return_value='10.0.0.1') - @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' - 'vm_active', return_value=True) - @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' - 'vm_ssh_active', return_value=True) - @mock.patch('snaps.openstack.create_instance.OpenStackVmInstance.' - 'ssh_client', return_value=True) - @mock.patch('scp.SCPClient') - @mock.patch('functest.opnfv_tests.openstack.vping.vping_ssh.' - 'VPingSSH._transfer_ping_script', return_value=True) - @mock.patch('functest.opnfv_tests.openstack.vping.vping_ssh.' - 'VPingSSH._do_vping_ssh', return_value=testcase.TestCase.EX_OK) - @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.' - 'get_ext_net_name', return_value='foo') - def test_vping_ssh(self, *args): - # pylint: disable=unused-argument - os_vm_inst = mock.MagicMock(name='get_console_output') - os_vm_inst.get_console_output.return_value = 'vPing OK' - ssh_client = mock.MagicMock(name='get_transport') - ssh_client.get_transport.return_value = None - scp_client = mock.MagicMock(name='put') - scp_client.put.return_value = None - - subnet_config = SubnetConfig( - name='bar', - cidr='10.0.0.1/24', - dns_nameservers=[env.get('NAMESERVER')]) - - with mock.patch('snaps.openstack.create_image.OpenStackImage.create', - return_value=OpenStackImage(self.os_creds, None)), \ - mock.patch( - 'snaps.openstack.create_network.OpenStackNetwork.create', - return_value=OpenStackNetwork( - self.os_creds, - NetworkConfig( - name='foo', - subnet_settings=[subnet_config]))), \ - mock.patch('snaps.openstack.utils.deploy_utils.' - 'create_vm_instance', - return_value=OpenStackVmInstance( - self.os_creds, - VmInstanceConfig( - name='foo', flavor='bar', - port_settings=[PortConfig( - name='foo', network_name='bar')]), - None)), \ - mock.patch('snaps.openstack.utils.deploy_utils.create_keypair', - return_value=OpenStackKeypair( - self.os_creds, KeypairConfig(name='foo'))), \ - mock.patch( - 'snaps.openstack.create_router.OpenStackRouter.create', - return_value=OpenStackRouter( - self.os_creds, RouterConfig(name='foo'))), \ - mock.patch('snaps.openstack.utils.deploy_utils.' - 'create_security_group', - return_value=OpenStackSecurityGroup( - self.os_creds, - SecurityGroupConfig(name='foo'))), \ - mock.patch('snaps.openstack.create_instance.' - 'OpenStackVmInstance.' - 'get_vm_inst', return_value=os_vm_inst), \ - mock.patch('snaps.openstack.create_instance.' - 'OpenStackVmInstance.' - 'ssh_client', return_value=ssh_client): - self.assertEquals(testcase.TestCase.EX_OK, self.vping_ssh.run()) - - -if __name__ == "__main__": - logging.disable(logging.CRITICAL) - unittest.main(verbosity=2) diff --git a/requirements.txt b/requirements.txt index 3196f52a8..493282c82 100644 --- a/requirements.txt +++ b/requirements.txt @@ -28,3 +28,5 @@ oslo.serialization!=2.19.1,>=1.10.0 # Apache-2.0 oslo.utils>=3.20.0 # Apache-2.0 xtesting refstack-client +os-client-config>=1.28.0 # Apache-2.0 +shade>=1.17.0 # Apache-2.0 |