X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=functest%2Fopnfv_tests%2Fopenstack%2Fvping%2Fvping_base.py;h=2451c2de4705e4ae040b7322c0c464154ba5d979;hb=9a78b4eb52ff9847971c14f6f92b43cb48a73f3d;hp=4390a99f73450aa39241ce30485e3e047b5b28c8;hpb=8c71f206875bab312072feeede1b9999efd76698;p=functest.git diff --git a/functest/opnfv_tests/openstack/vping/vping_base.py b/functest/opnfv_tests/openstack/vping/vping_base.py index 4390a99f7..2451c2de4 100644 --- a/functest/opnfv_tests/openstack/vping/vping_base.py +++ b/functest/opnfv_tests/openstack/vping/vping_base.py @@ -1,140 +1,147 @@ #!/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 # # http://www.apache.org/licenses/LICENSE-2.0 +"""Define the parent class of vping_ssh and vping_userdata testcases.""" + from datetime import datetime import logging -import os import time import uuid -from functest.core.testcase import TestCase -from functest.utils import functest_utils -from functest.utils.constants import CONST +import os_client_config +from xtesting.core import testcase -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 functest.utils import config +from functest.utils import env +from functest.utils import functest_utils -class VPingBase(TestCase): +class VPingBase(testcase.TestCase): """ Base class for vPing tests that check connectivity between two VMs shared internal network. This class is responsible for creating the image, internal network. """ + # pylint: disable=too-many-instance-attributes def __init__(self, **kwargs): super(VPingBase, self).__init__(**kwargs) - - self.logger = logging.getLogger(self.__class__.__name__) - - self.functest_repo = CONST.__getattribute__('dir_repo_functest') - self.guid = '' - if CONST.__getattribute__('vping_unique_names'): - self.guid = '-' + str(uuid.uuid4()) - - self.os_creds = openstack_tests.get_credentials( - os_env_file=CONST.__getattribute__('openstack_creds')) - - self.repo = CONST.__getattribute__('dir_vping') - - self.creators = list() - self.image_creator = None - self.network_creator = None - 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 - - # 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_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 = create_flavor.MEM_PAGE_SIZE_ANY - 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') - - def run(self): + self.logger = logging.getLogger(__name__) + self.cloud = os_client_config.make_shade() + self.ext_net = functest_utils.get_external_network(self.cloud) + self.logger.debug("ext_net: %s", self.ext_net) + self.guid = '-' + str(uuid.uuid4()) + self.network = None + self.subnet = None + self.router = None + self.image = None + self.flavor = None + self.vm1 = None + self.sec1 = None + + def run(self, **kwargs): # pylint: disable=too-many-locals """ 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) - + assert self.cloud + assert self.ext_net self.logger.info('Begin virtual environment setup') self.start_time = time.time() - self.logger.info("vPing Start Time:'%s'" % ( + self.logger.info( + "vPing Start Time:'%s'", 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) - 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.image_creator = deploy_utils.create_image( - self.os_creds, os_image_settings) - self.creators.append(self.image_creator) - + '%Y-%m-%d %H:%M:%S')) + + image_base_name = '{}-{}'.format( + getattr(config.CONF, 'vping_image_name'), self.guid) + self.logger.info("Creating image with name: '%s'", image_base_name) + meta = getattr(config.CONF, 'openstack_extra_properties', None) + self.logger.info("Image metadata: %s", meta) + self.image = self.cloud.create_image( + image_base_name, + filename=getattr(config.CONF, 'openstack_image_url'), + meta=meta) + self.logger.debug("image: %s", self.image) + + private_net_name = getattr( + config.CONF, 'vping_private_net_name') + self.guid + private_subnet_name = str(getattr( + config.CONF, 'vping_private_subnet_name') + self.guid) + private_subnet_cidr = getattr(config.CONF, 'vping_private_subnet_cidr') + + provider = {} + if hasattr(config.CONF, 'vping_network_type'): + provider["network_type"] = getattr( + config.CONF, 'vping_network_type') + if hasattr(config.CONF, 'vping_physical_network'): + provider["physical_network"] = getattr( + config.CONF, 'vping_physical_network') + if hasattr(config.CONF, 'vping_segmentation_id'): + provider["segmentation_id"] = getattr( + config.CONF, 'vping_segmentation_id') self.logger.info( - "Creating network with name: '%s'" % self.private_net_name) - self.network_creator = deploy_utils.create_network( - self.os_creds, - NetworkSettings(name=self.private_net_name, - subnet_settings=[SubnetSettings( - name=self.private_subnet_name, - cidr=self.private_subnet_cidr)])) - self.creators.append(self.network_creator) - + "Creating network with name: '%s'", private_net_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.logger.debug("subnet: %s", self.subnet) + + 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_name = 'vping-flavor' + self.guid self.logger.info( - "Creating flavor with name: '%s'" % self.flavor_name) - flavor_creator = OpenStackFlavor( - self.os_creds, - FlavorSettings(name=self.flavor_name, ram=512, disk=1, vcpus=1, - metadata=self.flavor_metadata)) - flavor_creator.create() - self.creators.append(flavor_creator) + "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', {})) + + self.sec1 = 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.sec1.id, protocol='icmp', direction='ingress') + + 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, + security_groups=[self.sec1.id]) + self.logger.debug("vm1: %s", self.vm1) + self.vm1 = self.cloud.wait_for_server(self.vm1, auto_ip=False) + p_console = self.cloud.get_server_console(self.vm1.id) + self.logger.debug("vm1 console: \n%s", p_console) def _execute(self): """ @@ -142,68 +149,32 @@ class VPingBase(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') - - if result == TestCase.EX_RUN_ERROR: - return TestCase.EX_RUN_ERROR - + 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.EX_OK + return testcase.TestCase.EX_OK - def _cleanup(self): + def clean(self): """ Cleanup all OpenStack objects. Should be called on completion :return: """ - if self.self_cleanup: - for creator in reversed(self.creators): - try: - creator.clean() - except Exception as e: - self.logger.error('Unexpected error cleaning - %s', e) - - def _do_vping(self, vm_creator, test_ip): + assert self.cloud + self.cloud.delete_server(self.vm1, wait=True) + self.cloud.delete_security_group(self.sec1.id) + 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') - - 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