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