X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=functest%2Fopnfv_tests%2Fopenstack%2Fvping%2Fvping_base.py;h=f3e0cfe326b121caf4c3fddf684c56110ade8e8a;hb=9a9856b49a1758501d11cb0a47c713f92c5adc11;hp=8e71bf82c2b22513cd42dd7bc130c62ebf3be947;hpb=9b36089c6a5f01e0770351d0e93b5f34410248a3;p=functest.git diff --git a/functest/opnfv_tests/openstack/vping/vping_base.py b/functest/opnfv_tests/openstack/vping/vping_base.py index 8e71bf82c..f3e0cfe32 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/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 @@ -13,7 +12,7 @@ import os import time import uuid -from functest.core.testcase import TestCase +from functest.core import testcase from functest.utils import functest_utils from functest.utils.constants import CONST @@ -21,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): +class VPingBase(testcase.TestCase): """ Base class for vPing tests that check connectivity between two VMs shared @@ -35,17 +34,17 @@ class VPingBase(TestCase): 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 = 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.logger = logging.getLogger(__name__) - self.repo = CONST.__getattribute__('dir_vping') + 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): 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 = 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') + 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,62 @@ class VPingBase(TestCase): 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') + + vping_network_type = None + vping_physical_network = None + vping_segmentation_id = None + + if (hasattr(CONST, 'network_type')): + vping_network_type = CONST.__getattribute__( + 'vping_network_type') + if (hasattr(CONST, 'physical_network')): + vping_physical_network = CONST.__getattribute__( + 'vping_physical_network') + if (hasattr(CONST, 'segmentation_id')): + vping_segmentation_id = CONST.__getattribute__( + 'vping_segmentation_id') + 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, - subnet_settings=[SubnetSettings( - name=self.private_subnet_name, - cidr=self.private_subnet_cidr)])) + NetworkSettings( + name=private_net_name, + network_type=vping_network_type, + physical_network=vping_physical_network, + segmentation_id=vping_segmentation_id, + subnet_settings=[SubnetSettings( + 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 = None + 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) @@ -152,19 +160,19 @@ class VPingBase(TestCase): else: raise Exception('VMs never became active') - if result == TestCase.EX_RUN_ERROR: - return TestCase.EX_RUN_ERROR + if result == testcase.TestCase.EX_RUN_ERROR: + return testcase.TestCase.EX_RUN_ERROR self.stop_time = time.time() self.result = 100 - return TestCase.EX_OK + return testcase.TestCase.EX_OK def _cleanup(self): """ 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 +188,3 @@ class VPingBase(TestCase): :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