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