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 functest.core import testcase
  19 from functest.opnfv_tests.openstack.snaps import snaps_utils
  20 from functest.utils.constants import CONST
  21
  22 from snaps.config.flavor import FlavorConfig
  23 from snaps.config.network import NetworkConfig, SubnetConfig
  24 from snaps.config.router import RouterConfig
  25 from snaps.openstack import create_flavor
  26 from snaps.openstack.create_flavor import OpenStackFlavor
  27 from snaps.openstack.tests import openstack_tests
  28 from snaps.openstack.utils import deploy_utils
  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         if 'os_creds' in kwargs:
  44             self.os_creds = kwargs['os_creds']
  45         else:
  46             creds_override = getattr(
  47                 CONST, 'snaps_os_creds_override') if hasattr(
  48                     CONST, 'snaps_os_creds_override') else None
  49             self.os_creds = openstack_tests.get_credentials(
  50                 os_env_file=getattr(CONST, 'openstack_creds'),
  51                 overrides=creds_override)
  52
  53         self.creators = list()
  54         self.image_creator = None
  55         self.network_creator = None
  56         self.vm1_creator = None
  57         self.vm2_creator = None
  58         self.router_creator = None
  59
  60         # Shared metadata
  61         self.guid = '-' + str(uuid.uuid4())
  62
  63         self.router_name = getattr(CONST, 'vping_router_name') + self.guid
  64         self.vm1_name = getattr(CONST, 'vping_vm_name_1') + self.guid
  65         self.vm2_name = getattr(CONST, 'vping_vm_name_2') + self.guid
  66
  67         self.vm_boot_timeout = getattr(CONST, 'vping_vm_boot_timeout')
  68         self.vm_delete_timeout = getattr(CONST, 'vping_vm_delete_timeout')
  69         self.vm_ssh_connect_timeout = getattr(
  70             CONST, 'vping_vm_ssh_connect_timeout')
  71         self.ping_timeout = getattr(CONST, 'vping_ping_timeout')
  72         self.flavor_name = 'vping-flavor' + self.guid
  73
  74         # Move this configuration option up for all tests to leverage
  75         if hasattr(CONST, 'snaps_images_cirros'):
  76             self.cirros_image_config = getattr(CONST, 'snaps_images_cirros')
  77         else:
  78             self.cirros_image_config = None
  79
  80     def run(self, **kwargs):  # pylint: disable=too-many-locals
  81         """
  82         Begins the test execution which should originate from the subclass
  83         """
  84         self.logger.info('Begin virtual environment setup')
  85
  86         self.start_time = time.time()
  87         self.logger.info(
  88             "vPing Start Time:'%s'",
  89             datetime.fromtimestamp(self.start_time).strftime(
  90                 '%Y-%m-%d %H:%M:%S'))
  91
  92         image_base_name = '{}-{}'.format(
  93             getattr(CONST, 'vping_image_name'),
  94             str(self.guid))
  95         os_image_settings = openstack_tests.cirros_image_settings(
  96             image_base_name, image_metadata=self.cirros_image_config)
  97         self.logger.info("Creating image with name: '%s'", image_base_name)
  98
  99         self.image_creator = deploy_utils.create_image(
 100             self.os_creds, os_image_settings)
 101         self.creators.append(self.image_creator)
 102
 103         private_net_name = getattr(CONST, 'vping_private_net_name') + self.guid
 104         private_subnet_name = getattr(
 105             CONST, 'vping_private_subnet_name') + self.guid
 106         private_subnet_cidr = getattr(CONST, 'vping_private_subnet_cidr')
 107
 108         vping_network_type = None
 109         vping_physical_network = None
 110         vping_segmentation_id = None
 111
 112         if hasattr(CONST, 'vping_network_type'):
 113             vping_network_type = getattr(CONST, 'vping_network_type')
 114         if hasattr(CONST, 'vping_physical_network'):
 115             vping_physical_network = getattr(CONST, 'vping_physical_network')
 116         if hasattr(CONST, 'vping_segmentation_id'):
 117             vping_segmentation_id = getattr(CONST, 'vping_segmentation_id')
 118
 119         self.logger.info(
 120             "Creating network with name: '%s'", private_net_name)
 121         self.network_creator = deploy_utils.create_network(
 122             self.os_creds,
 123             NetworkConfig(
 124                 name=private_net_name,
 125                 network_type=vping_network_type,
 126                 physical_network=vping_physical_network,
 127                 segmentation_id=vping_segmentation_id,
 128                 subnet_settings=[SubnetConfig(
 129                     name=private_subnet_name,
 130                     cidr=private_subnet_cidr)]))
 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 = getattr(CONST, '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(CONST, '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')