functest/opnfv_tests/openstack/vping/vping_base.py

   1 # Copyright (c) 2017 Cable Television Laboratories, Inc. and others.
   2 #
   3 # This program and the accompanying materials
   4 # are made available under the terms of the Apache License, Version 2.0
   5 # which accompanies this distribution, and is available at
   6 #
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8
   9 from datetime import datetime
  10 import logging
  11 import os
  12 import time
  13 import uuid
  14
  15 from functest.core import testcase
  16 from functest.opnfv_tests.openstack.snaps import snaps_utils
  17 from functest.utils.constants import CONST
  18
  19 from snaps.openstack import create_flavor
  20 from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
  21 from snaps.openstack.create_network import NetworkSettings, SubnetSettings
  22 from snaps.openstack.create_router import RouterSettings
  23 from snaps.openstack.tests import openstack_tests
  24 from snaps.openstack.utils import deploy_utils
  25
  26
  27 class VPingBase(testcase.TestCase):
  28
  29     """
  30     Base class for vPing tests that check connectivity between two VMs shared
  31     internal network.
  32     This class is responsible for creating the image, internal network.
  33     """
  34
  35     def __init__(self, **kwargs):
  36         super(VPingBase, self).__init__(**kwargs)
  37
  38         # This line is here simply for pep8 as the 'os' package import appears
  39         # to be required for mock and the unit tests will fail without it
  40         os.environ
  41
  42         self.logger = logging.getLogger(__name__)
  43
  44         if 'os_creds' in kwargs:
  45             self.os_creds = kwargs['os_creds']
  46         else:
  47             creds_override = None
  48             if hasattr(CONST, 'snaps_os_creds_override'):
  49                 creds_override = CONST.__getattribute__(
  50                     'snaps_os_creds_override')
  51
  52             self.os_creds = openstack_tests.get_credentials(
  53                 os_env_file=CONST.__getattribute__('openstack_creds'),
  54                 overrides=creds_override)
  55
  56         self.creators = list()
  57         self.image_creator = None
  58         self.network_creator = None
  59         self.vm1_creator = None
  60         self.vm2_creator = None
  61
  62         # Shared metadata
  63         self.guid = ''
  64         if CONST.__getattribute__('vping_unique_names'):
  65             self.guid = '-' + str(uuid.uuid4())
  66
  67         self.router_name = CONST.__getattribute__(
  68             'vping_router_name') + self.guid
  69         self.vm1_name = CONST.__getattribute__('vping_vm_name_1') + self.guid
  70         self.vm2_name = CONST.__getattribute__('vping_vm_name_2') + self.guid
  71
  72         self.vm_boot_timeout = CONST.__getattribute__('vping_vm_boot_timeout')
  73         self.vm_delete_timeout = CONST.__getattribute__(
  74             'vping_vm_delete_timeout')
  75         self.vm_ssh_connect_timeout = CONST.vping_vm_ssh_connect_timeout
  76         self.ping_timeout = CONST.__getattribute__('vping_ping_timeout')
  77         self.flavor_name = 'vping-flavor' + self.guid
  78
  79         # Move this configuration option up for all tests to leverage
  80         if hasattr(CONST, 'snaps_images_cirros'):
  81             self.cirros_image_config = CONST.__getattribute__(
  82                 'snaps_images_cirros')
  83         else:
  84             self.cirros_image_config = None
  85
  86     def run(self):
  87         """
  88         Begins the test execution which should originate from the subclass
  89         """
  90         self.logger.info('Begin virtual environment setup')
  91
  92         self.start_time = time.time()
  93         self.logger.info("vPing Start Time:'%s'" % (
  94             datetime.fromtimestamp(self.start_time).strftime(
  95                 '%Y-%m-%d %H:%M:%S')))
  96
  97         image_base_name = '{}-{}'.format(
  98             CONST.__getattribute__('vping_image_name'),
  99             str(self.guid))
 100         os_image_settings = openstack_tests.cirros_image_settings(
 101             image_base_name, image_metadata=self.cirros_image_config)
 102         self.logger.info("Creating image with name: '%s'" % image_base_name)
 103
 104         self.image_creator = deploy_utils.create_image(
 105             self.os_creds, os_image_settings)
 106         self.creators.append(self.image_creator)
 107
 108         private_net_name = CONST.__getattribute__(
 109             'vping_private_net_name') + self.guid
 110         private_subnet_name = CONST.__getattribute__(
 111             'vping_private_subnet_name') + self.guid
 112         private_subnet_cidr = CONST.__getattribute__(
 113             'vping_private_subnet_cidr')
 114
 115         vping_network_type = None
 116         vping_physical_network = None
 117         vping_segmentation_id = None
 118
 119         if (hasattr(CONST, 'vping_network_type')):
 120             vping_network_type = CONST.__getattribute__(
 121                 'vping_network_type')
 122         if (hasattr(CONST, 'vping_physical_network')):
 123             vping_physical_network = CONST.__getattribute__(
 124                 'vping_physical_network')
 125         if (hasattr(CONST, 'vping_segmentation_id')):
 126             vping_segmentation_id = CONST.__getattribute__(
 127                 'vping_segmentation_id')
 128
 129         self.logger.info(
 130             "Creating network with name: '%s'" % private_net_name)
 131         self.network_creator = deploy_utils.create_network(
 132             self.os_creds,
 133             NetworkSettings(
 134                 name=private_net_name,
 135                 network_type=vping_network_type,
 136                 physical_network=vping_physical_network,
 137                 segmentation_id=vping_segmentation_id,
 138                 subnet_settings=[SubnetSettings(
 139                     name=private_subnet_name,
 140                     cidr=private_subnet_cidr)]))
 141         self.creators.append(self.network_creator)
 142
 143         # Creating router to external network
 144         log = "Creating router with name: '%s'" % self.router_name
 145         self.logger.info(log)
 146         ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
 147         self.router_creator = deploy_utils.create_router(
 148             self.os_creds,
 149             RouterSettings(
 150                 name=self.router_name,
 151                 external_gateway=ext_net_name,
 152                 internal_subnets=[private_subnet_name]))
 153         self.creators.append(self.router_creator)
 154
 155         self.logger.info(
 156             "Creating flavor with name: '%s'" % self.flavor_name)
 157         scenario = CONST.__getattribute__('DEPLOY_SCENARIO')
 158         flavor_metadata = None
 159         if 'ovs' in scenario or 'fdio' in scenario:
 160             flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
 161         flavor_creator = OpenStackFlavor(
 162             self.os_creds,
 163             FlavorSettings(name=self.flavor_name, ram=512, disk=1, vcpus=1,
 164                            metadata=flavor_metadata))
 165         flavor_creator.create()
 166         self.creators.append(flavor_creator)
 167
 168     def _execute(self):
 169         """
 170         Method called by subclasses after environment has been setup
 171         :return: the exit code
 172         """
 173         self.logger.info('Begin test execution')
 174
 175         test_ip = self.vm1_creator.get_port_ip(
 176             self.vm1_creator.instance_settings.port_settings[0].name)
 177
 178         if self.vm1_creator.vm_active(
 179                 block=True) and self.vm2_creator.vm_active(block=True):
 180             result = self._do_vping(self.vm2_creator, test_ip)
 181         else:
 182             raise Exception('VMs never became active')
 183
 184         self.stop_time = time.time()
 185
 186         if result != testcase.TestCase.EX_OK:
 187             self.result = 0
 188             return testcase.TestCase.EX_RUN_ERROR
 189
 190         self.result = 100
 191         return testcase.TestCase.EX_OK
 192
 193     def _cleanup(self):
 194         """
 195         Cleanup all OpenStack objects. Should be called on completion
 196         :return:
 197         """
 198         if CONST.__getattribute__('vping_cleanup_objects'):
 199             for creator in reversed(self.creators):
 200                 try:
 201                     creator.clean()
 202                 except Exception as e:
 203                     self.logger.error('Unexpected error cleaning - %s', e)
 204
 205     def _do_vping(self, vm_creator, test_ip):
 206         """
 207         Method to be implemented by subclasses
 208         Begins the real test after the OpenStack environment has been setup
 209         :param vm_creator: the SNAPS VM instance creator object
 210         :param test_ip: the IP to which the VM needs to issue the ping
 211         :return: T/F
 212         """
 213         raise NotImplementedError('vping execution is not implemented')