functest/opnfv_tests/openstack/vping/vping_base.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 Cable Television Laboratories, Inc. and others.
   4 #
   5 # This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 #
   9 # http://www.apache.org/licenses/LICENSE-2.0
  10
  11 """Define the parent class of vping_ssh and vping_userdata testcases."""
  12
  13 from datetime import datetime
  14 import logging
  15 import time
  16 import uuid
  17
  18 from snaps.config.flavor import FlavorConfig
  19 from snaps.config.network import NetworkConfig, SubnetConfig
  20 from snaps.config.router import RouterConfig
  21 from snaps.openstack import create_flavor
  22 from snaps.openstack.create_flavor import OpenStackFlavor
  23 from snaps.openstack.tests import openstack_tests
  24 from snaps.openstack.utils import deploy_utils
  25 from xtesting.core import testcase
  26
  27 from functest.opnfv_tests.openstack.snaps import snaps_utils
  28 from functest.utils import config
  29 from functest.utils import env
  30
  31
  32 class VPingBase(testcase.TestCase):
  33
  34     """
  35     Base class for vPing tests that check connectivity between two VMs shared
  36     internal network.
  37     This class is responsible for creating the image, internal network.
  38     """
  39     # pylint: disable=too-many-instance-attributes
  40
  41     def __init__(self, **kwargs):
  42         super(VPingBase, self).__init__(**kwargs)
  43         self.logger = logging.getLogger(__name__)
  44         self.os_creds = kwargs.get('os_creds') or snaps_utils.get_credentials()
  45         self.creators = list()
  46         self.image_creator = None
  47         self.network_creator = None
  48         self.vm1_creator = None
  49         self.vm2_creator = None
  50         self.router_creator = None
  51
  52         # Shared metadata
  53         self.guid = '-' + str(uuid.uuid4())
  54
  55         self.router_name = getattr(
  56             config.CONF, 'vping_router_name') + self.guid
  57         self.vm1_name = getattr(
  58             config.CONF, 'vping_vm_name_1') + self.guid
  59         self.vm2_name = getattr(config.CONF, 'vping_vm_name_2') + self.guid
  60
  61         self.vm_boot_timeout = getattr(config.CONF, 'vping_vm_boot_timeout')
  62         self.vm_delete_timeout = getattr(
  63             config.CONF, 'vping_vm_delete_timeout')
  64         self.vm_ssh_connect_timeout = getattr(
  65             config.CONF, 'vping_vm_ssh_connect_timeout')
  66         self.ping_timeout = getattr(config.CONF, 'vping_ping_timeout')
  67         self.flavor_name = 'vping-flavor' + self.guid
  68
  69         # Move this configuration option up for all tests to leverage
  70         if hasattr(config.CONF, 'snaps_images_cirros'):
  71             self.cirros_image_config = getattr(
  72                 config.CONF, 'snaps_images_cirros')
  73         else:
  74             self.cirros_image_config = None
  75
  76     def run(self, **kwargs):  # pylint: disable=too-many-locals
  77         """
  78         Begins the test execution which should originate from the subclass
  79         """
  80         self.logger.info('Begin virtual environment setup')
  81
  82         self.start_time = time.time()
  83         self.logger.info(
  84             "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             getattr(config.CONF, '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 = getattr(
 100             config.CONF, 'vping_private_net_name') + self.guid
 101         private_subnet_name = getattr(
 102             config.CONF, 'vping_private_subnet_name') + self.guid
 103         private_subnet_cidr = getattr(config.CONF, 'vping_private_subnet_cidr')
 104
 105         vping_network_type = None
 106         vping_physical_network = None
 107         vping_segmentation_id = None
 108
 109         if hasattr(config.CONF, 'vping_network_type'):
 110             vping_network_type = getattr(config.CONF, 'vping_network_type')
 111         if hasattr(config.CONF, 'vping_physical_network'):
 112             vping_physical_network = getattr(
 113                 config.CONF, 'vping_physical_network')
 114         if hasattr(config.CONF, 'vping_segmentation_id'):
 115             vping_segmentation_id = getattr(
 116                 config.CONF, 'vping_segmentation_id')
 117
 118         self.logger.info(
 119             "Creating network with name: '%s'", private_net_name)
 120         self.network_creator = deploy_utils.create_network(
 121             self.os_creds,
 122             NetworkConfig(
 123                 name=private_net_name,
 124                 network_type=vping_network_type,
 125                 physical_network=vping_physical_network,
 126                 segmentation_id=vping_segmentation_id,
 127                 subnet_settings=[SubnetConfig(
 128                     name=private_subnet_name,
 129                     cidr=private_subnet_cidr,
 130                     dns_nameservers=[env.get('NAMESERVER')])]))
 131         self.creators.append(self.network_creator)
 132
 133         # Creating router to external network
 134         log = "Creating router with name: '%s'" % self.router_name
 135         self.logger.info(log)
 136         ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
 137         self.router_creator = deploy_utils.create_router(
 138             self.os_creds,
 139             RouterConfig(
 140                 name=self.router_name,
 141                 external_gateway=ext_net_name,
 142                 internal_subnets=[private_subnet_name]))
 143         self.creators.append(self.router_creator)
 144
 145         self.logger.info(
 146             "Creating flavor with name: '%s'", self.flavor_name)
 147         scenario = env.get('DEPLOY_SCENARIO')
 148         flavor_metadata = None
 149         flavor_ram = 512
 150         if 'ovs' in scenario or 'fdio' in scenario:
 151             flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
 152             flavor_ram = 1024
 153         flavor_creator = OpenStackFlavor(
 154             self.os_creds,
 155             FlavorConfig(name=self.flavor_name, ram=flavor_ram, disk=1,
 156                          vcpus=1, metadata=flavor_metadata))
 157         flavor_creator.create()
 158         self.creators.append(flavor_creator)
 159
 160     def _execute(self):
 161         """
 162         Method called by subclasses after environment has been setup
 163         :return: the exit code
 164         """
 165         self.logger.info('Begin test execution')
 166
 167         test_ip = self.vm1_creator.get_port_ip(
 168             self.vm1_creator.instance_settings.port_settings[0].name)
 169
 170         if self.vm1_creator.vm_active(
 171                 block=True) and self.vm2_creator.vm_active(block=True):
 172             result = self._do_vping(self.vm2_creator, test_ip)
 173         else:
 174             raise Exception('VMs never became active')
 175
 176         self.stop_time = time.time()
 177
 178         if result != testcase.TestCase.EX_OK:
 179             self.result = 0
 180             return testcase.TestCase.EX_RUN_ERROR
 181
 182         self.result = 100
 183         return testcase.TestCase.EX_OK
 184
 185     def _cleanup(self):
 186         """
 187         Cleanup all OpenStack objects. Should be called on completion
 188         :return:
 189         """
 190         if getattr(config.CONF, 'vping_cleanup_objects') == 'True':
 191             for creator in reversed(self.creators):
 192                 try:
 193                     creator.clean()
 194                 except Exception as error:  # pylint: disable=broad-except
 195                     self.logger.error('Unexpected error cleaning - %s', error)
 196
 197     def _do_vping(self, vm_creator, test_ip):
 198         """
 199         Method to be implemented by subclasses
 200         Begins the real test after the OpenStack environment has been setup
 201         :param vm_creator: the SNAPS VM instance creator object
 202         :param test_ip: the IP to which the VM needs to issue the ping
 203         :return: T/F
 204         """
 205         raise NotImplementedError('vping execution is not implemented')