functest/opnfv_tests/openstack/vping/vping_base.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright (c) 2015 All rights reserved
   4 # This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 #
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 from datetime import datetime
  11 import logging
  12 import os
  13 import time
  14 import uuid
  15
  16 from functest.core.testcase import TestCase
  17 from functest.utils import functest_utils
  18 from functest.utils.constants import CONST
  19
  20 from snaps.openstack import create_flavor
  21 from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
  22 from snaps.openstack.create_network import NetworkSettings, SubnetSettings
  23 from snaps.openstack.tests import openstack_tests
  24 from snaps.openstack.utils import deploy_utils, nova_utils
  25
  26
  27 class VPingBase(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     def __init__(self, **kwargs):
  35         super(VPingBase, self).__init__(**kwargs)
  36
  37         self.logger = logging.getLogger(self.__class__.__name__)
  38
  39         self.functest_repo = CONST.__getattribute__('dir_repo_functest')
  40         self.guid = ''
  41         if CONST.__getattribute__('vping_unique_names'):
  42             self.guid = '-' + str(uuid.uuid4())
  43
  44         self.os_creds = openstack_tests.get_credentials(
  45             os_env_file=CONST.__getattribute__('openstack_creds'))
  46
  47         self.repo = CONST.__getattribute__('dir_vping')
  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         self.self_cleanup = CONST.__getattribute__('vping_cleanup_objects')
  56
  57         # Image constants
  58         self.image_name =\
  59             CONST.__getattribute__('vping_image_name') + self.guid
  60
  61         # VM constants
  62         self.vm1_name = CONST.__getattribute__('vping_vm_name_1') + self.guid
  63         self.vm2_name = CONST.__getattribute__('vping_vm_name_2') + self.guid
  64         self.vm_boot_timeout = CONST.__getattribute__('vping_vm_boot_timeout')
  65         self.vm_delete_timeout =\
  66             CONST.__getattribute__('vping_vm_delete_timeout')
  67         self.vm_ssh_connect_timeout = CONST.vping_vm_ssh_connect_timeout
  68         self.ping_timeout = CONST.__getattribute__('vping_ping_timeout')
  69         self.flavor_name = 'vping-flavor' + self.guid
  70
  71         # NEUTRON Private Network parameters
  72         self.private_net_name =\
  73             CONST.__getattribute__('vping_private_net_name') + self.guid
  74         self.private_subnet_name =\
  75             CONST.__getattribute__('vping_private_subnet_name') + self.guid
  76         self.private_subnet_cidr =\
  77             CONST.__getattribute__('vping_private_subnet_cidr')
  78
  79         scenario = functest_utils.get_scenario()
  80
  81         self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_ANY
  82         if 'ovs' in scenario or 'fdio' in scenario:
  83             self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
  84
  85         self.cirros_image_config = None
  86
  87         # Move this configuration option up for all tests to leverage
  88         if hasattr(CONST, 'snaps_images_cirros'):
  89             self.cirros_image_config = CONST.__getattribute__(
  90                 'snaps_images_cirros')
  91
  92     def run(self):
  93         """
  94         Begins the test execution which should originate from the subclass
  95         """
  96
  97         if not os.path.exists(self.functest_repo):
  98             raise Exception(
  99                 "Functest repository not found '%s'" % self.functest_repo)
 100
 101         self.logger.info('Begin virtual environment setup')
 102
 103         self.start_time = time.time()
 104         self.logger.info("vPing Start Time:'%s'" % (
 105             datetime.fromtimestamp(self.start_time).strftime(
 106                 '%Y-%m-%d %H:%M:%S')))
 107
 108         self.__delete_exist_vms()
 109
 110         image_base_name = self.image_name + '-' + str(self.guid)
 111         os_image_settings = openstack_tests.cirros_image_settings(
 112             image_base_name, image_metadata=self.cirros_image_config)
 113         self.logger.info("Creating image with name: '%s'" % self.image_name)
 114
 115         self.image_creator = deploy_utils.create_image(
 116             self.os_creds, os_image_settings)
 117         self.creators.append(self.image_creator)
 118
 119         self.logger.info(
 120             "Creating network with name: '%s'" % self.private_net_name)
 121         self.network_creator = deploy_utils.create_network(
 122             self.os_creds,
 123             NetworkSettings(name=self.private_net_name,
 124                             subnet_settings=[SubnetSettings(
 125                                 name=self.private_subnet_name,
 126                                 cidr=self.private_subnet_cidr)]))
 127         self.creators.append(self.network_creator)
 128
 129         self.logger.info(
 130             "Creating flavor with name: '%s'" % self.flavor_name)
 131         flavor_creator = OpenStackFlavor(
 132             self.os_creds,
 133             FlavorSettings(name=self.flavor_name, ram=512, disk=1, vcpus=1,
 134                            metadata=self.flavor_metadata))
 135         flavor_creator.create()
 136         self.creators.append(flavor_creator)
 137
 138     def _execute(self):
 139         """
 140         Method called by subclasses after environment has been setup
 141         :return: the exit code
 142         """
 143         self.logger.info('Begin test execution')
 144
 145         test_ip = self.vm1_creator.get_port_ip(
 146             self.vm1_creator.instance_settings.port_settings[0].name)
 147
 148         if self.vm1_creator.vm_active(
 149                 block=True) and self.vm2_creator.vm_active(block=True):
 150             result = self._do_vping(self.vm2_creator, test_ip)
 151         else:
 152             raise Exception('VMs never became active')
 153
 154         if result == TestCase.EX_RUN_ERROR:
 155             return TestCase.EX_RUN_ERROR
 156
 157         self.stop_time = time.time()
 158         self.result = 'PASS'
 159         return TestCase.EX_OK
 160
 161     def _cleanup(self):
 162         """
 163         Cleanup all OpenStack objects. Should be called on completion
 164         :return:
 165         """
 166         if self.self_cleanup:
 167             for creator in reversed(self.creators):
 168                 try:
 169                     creator.clean()
 170                 except Exception as e:
 171                     self.logger.error('Unexpected error cleaning - %s', e)
 172
 173     def _do_vping(self, vm_creator, test_ip):
 174         """
 175         Method to be implemented by subclasses
 176         Begins the real test after the OpenStack environment has been setup
 177         :param vm_creator: the SNAPS VM instance creator object
 178         :param test_ip: the IP to which the VM needs to issue the ping
 179         :return: T/F
 180         """
 181         raise NotImplementedError('vping execution is not implemented')
 182
 183     def __delete_exist_vms(self):
 184         """
 185         Cleans any existing VMs using the same name.
 186         """
 187         nova_client = nova_utils.nova_client(self.os_creds)
 188         servers = nova_client.servers.list()
 189         for server in servers:
 190             if server.name == self.vm1_name or server.name == self.vm2_name:
 191                 self.logger.info("Deleting instance %s..." % server.name)
 192                 server.delete()
 193
 194
 195 class VPingMain(object):
 196     def __init__(self, vping_cls):
 197         self.vping = vping_cls()
 198
 199     def main(self, **kwargs):
 200         try:
 201             result = self.vping.run(**kwargs)
 202             if result != VPingBase.EX_OK:
 203                 return result
 204             if kwargs['report']:
 205                 return self.vping.publish_report()
 206         except:
 207             return VPingBase.EX_RUN_ERROR