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