diff options
| -rw-r--r-- | functest/opnfv_tests/openstack/vping/vping_base.py | 120 | ||||
| -rwxr-xr-x | functest/opnfv_tests/openstack/vping/vping_ssh.py | 36 | ||||
| -rw-r--r-- | functest/tests/unit/openstack/vping/test_vping.py | 157 |
3 files changed, 213 insertions, 100 deletions
diff --git a/functest/opnfv_tests/openstack/vping/vping_base.py b/functest/opnfv_tests/openstack/vping/vping_base.py index 714606769..7de16ce15 100644 --- a/functest/opnfv_tests/openstack/vping/vping_base.py +++ b/functest/opnfv_tests/openstack/vping/vping_base.py @@ -1,6 +1,5 @@ -#!/usr/bin/env python +# Copyright (c) 2017 Cable Television Laboratories, Inc. and others. # -# Copyright (c) 2015 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 @@ -10,7 +9,6 @@ from datetime import datetime import logging import os -import pkg_resources import time import uuid @@ -22,10 +20,10 @@ from snaps.openstack import create_flavor from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor from snaps.openstack.create_network import NetworkSettings, SubnetSettings from snaps.openstack.tests import openstack_tests -from snaps.openstack.utils import deploy_utils, nova_utils +from snaps.openstack.utils import deploy_utils -class VPingBase(testcase.OSGCTestCase): +class VPingBase(testcase.TestCase): """ Base class for vPing tests that check connectivity between two VMs shared @@ -36,16 +34,17 @@ class VPingBase(testcase.OSGCTestCase): def __init__(self, **kwargs): super(VPingBase, self).__init__(**kwargs) - self.logger = logging.getLogger(self.__class__.__name__) + # This line is here simply for pep8 as the 'os' package import appears + # to be required for mock and the unit tests will fail without it + os.environ - self.functest_repo = pkg_resources.resource_filename( - 'functest', '..') - self.guid = '' - if CONST.__getattribute__('vping_unique_names'): - self.guid = '-' + str(uuid.uuid4()) + self.logger = logging.getLogger(self.__class__.__name__) - self.os_creds = openstack_tests.get_credentials( - os_env_file=CONST.__getattribute__('openstack_creds')) + if 'os_creds' in kwargs: + self.os_creds = kwargs['os_creds'] + else: + self.os_creds = openstack_tests.get_credentials( + os_env_file=CONST.__getattribute__('openstack_creds')) self.creators = list() self.image_creator = None @@ -53,52 +52,32 @@ class VPingBase(testcase.OSGCTestCase): self.vm1_creator = None self.vm2_creator = None - self.self_cleanup = CONST.__getattribute__('vping_cleanup_objects') - - # Image constants - self.image_name =\ - CONST.__getattribute__('vping_image_name') + self.guid + # Shared metadata + self.guid = '' + if CONST.__getattribute__('vping_unique_names'): + self.guid = '-' + str(uuid.uuid4()) - # VM constants self.vm1_name = CONST.__getattribute__('vping_vm_name_1') + self.guid self.vm2_name = CONST.__getattribute__('vping_vm_name_2') + self.guid + self.vm_boot_timeout = CONST.__getattribute__('vping_vm_boot_timeout') - self.vm_delete_timeout =\ - CONST.__getattribute__('vping_vm_delete_timeout') + self.vm_delete_timeout = CONST.__getattribute__( + 'vping_vm_delete_timeout') self.vm_ssh_connect_timeout = CONST.vping_vm_ssh_connect_timeout self.ping_timeout = CONST.__getattribute__('vping_ping_timeout') self.flavor_name = 'vping-flavor' + self.guid - # NEUTRON Private Network parameters - self.private_net_name =\ - CONST.__getattribute__('vping_private_net_name') + self.guid - self.private_subnet_name =\ - CONST.__getattribute__('vping_private_subnet_name') + self.guid - self.private_subnet_cidr =\ - CONST.__getattribute__('vping_private_subnet_cidr') - - scenario = functest_utils.get_scenario() - - self.flavor_metadata = None - if 'ovs' in scenario or 'fdio' in scenario: - self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE - - self.cirros_image_config = None - # Move this configuration option up for all tests to leverage if hasattr(CONST, 'snaps_images_cirros'): self.cirros_image_config = CONST.__getattribute__( 'snaps_images_cirros') + else: + self.cirros_image_config = None def run(self): """ Begins the test execution which should originate from the subclass """ - - if not os.path.exists(self.functest_repo): - raise Exception( - "Functest repository not found '%s'" % self.functest_repo) - self.logger.info('Begin virtual environment setup') self.start_time = time.time() @@ -106,33 +85,43 @@ class VPingBase(testcase.OSGCTestCase): datetime.fromtimestamp(self.start_time).strftime( '%Y-%m-%d %H:%M:%S'))) - self.__delete_exist_vms() - - image_base_name = self.image_name + '-' + str(self.guid) + image_base_name = '{}-{}'.format( + CONST.__getattribute__('vping_image_name'), + str(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'" % self.image_name) + self.logger.info("Creating image with name: '%s'" % image_base_name) self.image_creator = deploy_utils.create_image( self.os_creds, os_image_settings) self.creators.append(self.image_creator) + private_net_name = CONST.__getattribute__( + 'vping_private_net_name') + self.guid + private_subnet_name = CONST.__getattribute__( + 'vping_private_subnet_name') + self.guid + private_subnet_cidr = CONST.__getattribute__( + 'vping_private_subnet_cidr') self.logger.info( - "Creating network with name: '%s'" % self.private_net_name) + "Creating network with name: '%s'" % private_net_name) self.network_creator = deploy_utils.create_network( self.os_creds, - NetworkSettings(name=self.private_net_name, + NetworkSettings(name=private_net_name, subnet_settings=[SubnetSettings( - name=self.private_subnet_name, - cidr=self.private_subnet_cidr)])) + name=private_subnet_name, + cidr=private_subnet_cidr)])) self.creators.append(self.network_creator) self.logger.info( "Creating flavor with name: '%s'" % self.flavor_name) + scenario = functest_utils.get_scenario() + flavor_metadata = create_flavor.MEM_PAGE_SIZE_ANY + if 'ovs' in scenario or 'fdio' in scenario: + flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE flavor_creator = OpenStackFlavor( self.os_creds, FlavorSettings(name=self.flavor_name, ram=512, disk=1, vcpus=1, - metadata=self.flavor_metadata)) + metadata=flavor_metadata)) flavor_creator.create() self.creators.append(flavor_creator) @@ -164,7 +153,7 @@ class VPingBase(testcase.OSGCTestCase): Cleanup all OpenStack objects. Should be called on completion :return: """ - if self.self_cleanup: + if CONST.__getattribute__('vping_cleanup_objects'): for creator in reversed(self.creators): try: creator.clean() @@ -180,30 +169,3 @@ class VPingBase(testcase.OSGCTestCase): :return: T/F """ raise NotImplementedError('vping execution is not implemented') - - def __delete_exist_vms(self): - """ - Cleans any existing VMs using the same name. - """ - nova_client = nova_utils.nova_client(self.os_creds) - servers = nova_client.servers.list() - for server in servers: - if server.name == self.vm1_name or server.name == self.vm2_name: - self.logger.info("Deleting instance %s..." % server.name) - server.delete() - - -class VPingMain(object): - - def __init__(self, vping_cls): - self.vping = vping_cls() - - def main(self, **kwargs): - try: - result = self.vping.run(**kwargs) - if result != VPingBase.EX_OK: - return result - if kwargs['report']: - return self.vping.publish_report() - except: - return VPingBase.EX_RUN_ERROR diff --git a/functest/opnfv_tests/openstack/vping/vping_ssh.py b/functest/opnfv_tests/openstack/vping/vping_ssh.py index 1f663307c..eacccb988 100755 --- a/functest/opnfv_tests/openstack/vping/vping_ssh.py +++ b/functest/opnfv_tests/openstack/vping/vping_ssh.py @@ -7,10 +7,9 @@ # # http://www.apache.org/licenses/LICENSE-2.0 -import argparse +import os import pkg_resources from scp import SCPClient -import sys import time from snaps.openstack.create_instance import FloatingIpSettings, \ @@ -35,6 +34,11 @@ class VPingSSH(vping_base.VPingBase): """ def __init__(self, **kwargs): + + # This line is here simply for pep8 as the 'os' package import appears + # to be required for mock and the unit tests will fail without it + os.environ + if "case_name" not in kwargs: kwargs["case_name"] = "vping_ssh" super(VPingSSH, self).__init__(**kwargs) @@ -42,13 +46,11 @@ class VPingSSH(vping_base.VPingBase): self.kp_name = CONST.__getattribute__('vping_keypair_name') + self.guid self.kp_priv_file = CONST.__getattribute__('vping_keypair_priv_file') self.kp_pub_file = CONST.__getattribute__('vping_keypair_pub_file') - self.router_name =\ - CONST.__getattribute__('vping_router_name') + self.guid + self.router_name = CONST.__getattribute__( + 'vping_router_name') + self.guid self.sg_name = CONST.__getattribute__('vping_sg_name') + self.guid self.sg_desc = CONST.__getattribute__('vping_sg_desc') - self.ext_net_name = snaps_utils.get_ext_net_name(self.os_creds) - def run(self): """ Sets up the OpenStack keypair, router, security group, and VM instance @@ -71,11 +73,12 @@ class VPingSSH(vping_base.VPingBase): % self.router_name) net_set = self.network_creator.network_settings sub_set = [net_set.subnet_settings[0].name] + ext_net_name = snaps_utils.get_ext_net_name(self.os_creds) router_creator = deploy_utils.create_router( self.os_creds, RouterSettings( name=self.router_name, - external_gateway=self.ext_net_name, + external_gateway=ext_net_name, internal_subnets=sub_set)) self.creators.append(router_creator) @@ -142,13 +145,13 @@ class VPingSSH(vping_base.VPingBase): """ if vm_creator.vm_ssh_active(block=True): ssh = vm_creator.ssh_client() - if not self.__transfer_ping_script(ssh): + if not self._transfer_ping_script(ssh): return TestCase.EX_RUN_ERROR - return self.__do_vping_ssh(ssh, test_ip) + return self._do_vping_ssh(ssh, test_ip) else: return -1 - def __transfer_ping_script(self, ssh): + def _transfer_ping_script(self, ssh): """ Uses SCP to copy the ping script via the SSH client :param ssh: the SSH client @@ -157,7 +160,7 @@ class VPingSSH(vping_base.VPingBase): self.logger.info("Trying to transfer ping.sh") scp = SCPClient(ssh.get_transport()) ping_script = pkg_resources.resource_filename( - 'functest', 'opnfv_tests/openstack/vping/ping.sh') + 'functest.opnfv_tests.openstack.vping', 'ping.sh') try: scp.put(ping_script, "~/") except: @@ -171,7 +174,7 @@ class VPingSSH(vping_base.VPingBase): return True - def __do_vping_ssh(self, ssh, test_ip): + def _do_vping_ssh(self, ssh, test_ip): """ Pings the test_ip via the SSH client :param ssh: the SSH client used to issue the ping command @@ -234,12 +237,3 @@ class VPingSSH(vping_base.VPingBase): name=self.sg_name, description=self.sg_desc, rule_settings=sg_rules)) - - -if __name__ == '__main__': - args_parser = argparse.ArgumentParser() - args_parser.add_argument("-r", "--report", - help="Create json result file", - action="store_true") - args = vars(args_parser.parse_args()) - sys.exit(vping_base.VPingMain(VPingSSH).main(**args)) diff --git a/functest/tests/unit/openstack/vping/test_vping.py b/functest/tests/unit/openstack/vping/test_vping.py new file mode 100644 index 000000000..b229c3519 --- /dev/null +++ b/functest/tests/unit/openstack/vping/test_vping.py @@ -0,0 +1,157 @@ +# 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 + +import unittest + +import mock + +from snaps.openstack.create_image import OpenStackImage +from snaps.openstack.create_instance import OpenStackVmInstance, \ + VmInstanceSettings +from snaps.openstack.create_keypairs import OpenStackKeypair, KeypairSettings +from snaps.openstack.create_network import OpenStackNetwork, NetworkSettings, \ + SubnetSettings, PortSettings +from snaps.openstack.create_router import OpenStackRouter, RouterSettings +from snaps.openstack.create_security_group import OpenStackSecurityGroup, \ + SecurityGroupSettings +from snaps.openstack.os_credentials import OSCreds + +from functest.core.testcase import TestCase +from functest.opnfv_tests.openstack.vping import vping_userdata, vping_ssh + + +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('functest.opnfv_tests.openstack.vping.vping_base.os.' + 'path.exists', return_value=True) + @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) + def test_vping_userdata(self, deploy_vm, path_exists, create_flavor, + get_port_ip, vm_active): + os_vm_inst = mock.MagicMock(name='get_console_output') + os_vm_inst.get_console_output.return_value = 'vPing OK' + with mock.patch('snaps.openstack.utils.deploy_utils.create_image', + return_value=OpenStackImage(self.os_creds, None)), \ + mock.patch('snaps.openstack.utils.deploy_utils.create_network', + return_value=OpenStackNetwork( + self.os_creds, NetworkSettings(name='foo'))), \ + mock.patch('snaps.openstack.utils.deploy_utils.' + 'create_vm_instance', + return_value=OpenStackVmInstance( + self.os_creds, + VmInstanceSettings( + name='foo', flavor='bar', + port_settings=[PortSettings( + name='foo', network_name='bar')]), + None)), \ + mock.patch('snaps.openstack.create_instance.' + 'OpenStackVmInstance.get_os_vm_server_obj', + return_value=os_vm_inst): + self.assertEquals(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('functest.opnfv_tests.openstack.vping.vping_base.os.' + 'path.exists', return_value=True) + @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.EX_OK) + @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.' + 'get_ext_net_name', return_value='foo') + def test_vping_ssh(self, create_vm, path_exists, + flavor_create, get_port_ip, vm_active, ssh_active, + ssh_client, scp_client, trans_script, do_vping_ssh, + ext_net_name): + 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 + + with mock.patch('snaps.openstack.utils.deploy_utils.create_image', + return_value=OpenStackImage(self.os_creds, None)), \ + mock.patch('snaps.openstack.utils.deploy_utils.create_network', + return_value=OpenStackNetwork( + self.os_creds, + NetworkSettings( + name='foo', + subnet_settings=[ + SubnetSettings( + name='bar', + cidr='10.0.0.1/24')]))), \ + mock.patch('snaps.openstack.utils.deploy_utils.' + 'create_vm_instance', + return_value=OpenStackVmInstance( + self.os_creds, + VmInstanceSettings( + name='foo', flavor='bar', + port_settings=[PortSettings( + name='foo', network_name='bar')]), + None)), \ + mock.patch('snaps.openstack.utils.deploy_utils.create_keypair', + return_value=OpenStackKeypair( + self.os_creds, KeypairSettings(name='foo'))), \ + mock.patch('snaps.openstack.utils.deploy_utils.create_router', + return_value=OpenStackRouter( + self.os_creds, RouterSettings(name='foo'))), \ + mock.patch('snaps.openstack.utils.deploy_utils.' + 'create_security_group', + return_value=OpenStackSecurityGroup( + self.os_creds, + SecurityGroupSettings(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.EX_OK, self.vping_ssh.run()) |