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