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