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(self.__class__.__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         self.logger.info(
 106             "Creating network with name: '%s'" % private_net_name)
 107         self.network_creator = deploy_utils.create_network(
 108             self.os_creds,
 109             NetworkSettings(name=private_net_name,
 110                             subnet_settings=[SubnetSettings(
 111                                 name=private_subnet_name,
 112                                 cidr=private_subnet_cidr)]))
 113         self.creators.append(self.network_creator)
 114
 115         self.logger.info(
 116             "Creating flavor with name: '%s'" % self.flavor_name)
 117         scenario = functest_utils.get_scenario()
 118         flavor_metadata = None
 119         if 'ovs' in scenario or 'fdio' in scenario:
 120             flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
 121         flavor_creator = OpenStackFlavor(
 122             self.os_creds,
 123             FlavorSettings(name=self.flavor_name, ram=512, disk=1, vcpus=1,
 124                            metadata=flavor_metadata))
 125         flavor_creator.create()
 126         self.creators.append(flavor_creator)
 127
 128     def _execute(self):
 129         """
 130         Method called by subclasses after environment has been setup
 131         :return: the exit code
 132         """
 133         self.logger.info('Begin test execution')
 134
 135         test_ip = self.vm1_creator.get_port_ip(
 136             self.vm1_creator.instance_settings.port_settings[0].name)
 137
 138         if self.vm1_creator.vm_active(
 139                 block=True) and self.vm2_creator.vm_active(block=True):
 140             result = self._do_vping(self.vm2_creator, test_ip)
 141         else:
 142             raise Exception('VMs never became active')
 143
 144         if result == testcase.TestCase.EX_RUN_ERROR:
 145             return testcase.TestCase.EX_RUN_ERROR
 146
 147         self.stop_time = time.time()
 148         self.result = 100
 149         return testcase.TestCase.EX_OK
 150
 151     def _cleanup(self):
 152         """
 153         Cleanup all OpenStack objects. Should be called on completion
 154         :return:
 155         """
 156         if CONST.__getattribute__('vping_cleanup_objects'):
 157             for creator in reversed(self.creators):
 158                 try:
 159                     creator.clean()
 160                 except Exception as e:
 161                     self.logger.error('Unexpected error cleaning - %s', e)
 162
 163     def _do_vping(self, vm_creator, test_ip):
 164         """
 165         Method to be implemented by subclasses
 166         Begins the real test after the OpenStack environment has been setup
 167         :param vm_creator: the SNAPS VM instance creator object
 168         :param test_ip: the IP to which the VM needs to issue the ping
 169         :return: T/F
 170         """
 171         raise NotImplementedError('vping execution is not implemented')