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 import os_client_config
  19 from xtesting.core import testcase
  20
  21 from functest.utils import config
  22 from functest.utils import env
  23
  24
  25 class VPingBase(testcase.TestCase):
  26
  27     """
  28     Base class for vPing tests that check connectivity between two VMs shared
  29     internal network.
  30     This class is responsible for creating the image, internal network.
  31     """
  32     # pylint: disable=too-many-instance-attributes
  33
  34     def __init__(self, **kwargs):
  35         super(VPingBase, self).__init__(**kwargs)
  36         self.logger = logging.getLogger(__name__)
  37         self.cloud = os_client_config.make_shade()
  38         self.ext_net = self.cloud.get_network("ext-net")
  39         self.logger.debug("ext_net: %s", self.ext_net)
  40         self.guid = '-' + str(uuid.uuid4())
  41         self.network = None
  42         self.subnet = None
  43         self.router = None
  44         self.image = None
  45         self.flavor = None
  46         self.vm1 = None
  47
  48     def run(self, **kwargs):  # pylint: disable=too-many-locals
  49         """
  50         Begins the test execution which should originate from the subclass
  51         """
  52         assert self.cloud
  53         assert self.ext_net
  54         self.logger.info('Begin virtual environment setup')
  55
  56         self.start_time = time.time()
  57         self.logger.info(
  58             "vPing Start Time:'%s'",
  59             datetime.fromtimestamp(self.start_time).strftime(
  60                 '%Y-%m-%d %H:%M:%S'))
  61
  62         image_base_name = '{}-{}'.format(
  63             getattr(config.CONF, 'vping_image_name'), self.guid)
  64         self.logger.info("Creating image with name: '%s'", image_base_name)
  65         self.image = self.cloud.create_image(
  66             image_base_name,
  67             filename=getattr(config.CONF, 'openstack_image_url'))
  68         self.logger.debug("image: %s", self.image)
  69
  70         private_net_name = getattr(
  71             config.CONF, 'vping_private_net_name') + self.guid
  72         private_subnet_name = str(getattr(
  73             config.CONF, 'vping_private_subnet_name') + self.guid)
  74         private_subnet_cidr = getattr(config.CONF, 'vping_private_subnet_cidr')
  75
  76         provider = {}
  77         if hasattr(config.CONF, 'vping_network_type'):
  78             provider["network_type"] = getattr(
  79                 config.CONF, 'vping_network_type')
  80         if hasattr(config.CONF, 'vping_physical_network'):
  81             provider["physical_network"] = getattr(
  82                 config.CONF, 'vping_physical_network')
  83         if hasattr(config.CONF, 'vping_segmentation_id'):
  84             provider["segmentation_id"] = getattr(
  85                 config.CONF, 'vping_segmentation_id')
  86         self.logger.info(
  87             "Creating network with name: '%s'", private_net_name)
  88         self.network = self.cloud.create_network(
  89             private_net_name,
  90             provider=provider)
  91         self.logger.debug("network: %s", self.network)
  92
  93         self.subnet = self.cloud.create_subnet(
  94             self.network.id,
  95             subnet_name=private_subnet_name,
  96             cidr=private_subnet_cidr,
  97             enable_dhcp=True,
  98             dns_nameservers=[env.get('NAMESERVER')])
  99         self.logger.debug("subnet: %s", self.subnet)
 100
 101         router_name = getattr(config.CONF, 'vping_router_name') + self.guid
 102         self.logger.info("Creating router with name: '%s'", router_name)
 103         self.router = self.cloud.create_router(
 104             name=router_name,
 105             ext_gateway_net_id=self.ext_net.id)
 106         self.logger.debug("router: %s", self.router)
 107         self.cloud.add_router_interface(self.router, subnet_id=self.subnet.id)
 108
 109         flavor_name = 'vping-flavor' + self.guid
 110         self.logger.info(
 111             "Creating flavor with name: '%s'", flavor_name)
 112         self.flavor = self.cloud.create_flavor(
 113             flavor_name, getattr(config.CONF, 'openstack_flavor_ram'),
 114             getattr(config.CONF, 'openstack_flavor_vcpus'),
 115             getattr(config.CONF, 'openstack_flavor_disk'))
 116         self.logger.debug("flavor: %s", self.flavor)
 117         self.cloud.set_flavor_specs(
 118             self.flavor.id, getattr(config.CONF, 'flavor_extra_specs', {}))
 119
 120         vm1_name = getattr(config.CONF, 'vping_vm_name_1') + self.guid
 121         self.logger.info(
 122             "Creating VM 1 instance with name: '%s'", vm1_name)
 123         self.vm1 = self.cloud.create_server(
 124             vm1_name, image=self.image.id,
 125             flavor=self.flavor.id,
 126             auto_ip=False, wait=True,
 127             timeout=getattr(config.CONF, 'vping_vm_boot_timeout'),
 128             network=self.network.id)
 129         self.logger.debug("vm1: %s", self.vm1)
 130         self.vm1 = self.cloud.wait_for_server(self.vm1, auto_ip=False)
 131
 132     def _execute(self):
 133         """
 134         Method called by subclasses after environment has been setup
 135         :return: the exit code
 136         """
 137         self.logger.info('Begin test execution')
 138         result = self._do_vping()
 139         self.stop_time = time.time()
 140         if result != testcase.TestCase.EX_OK:
 141             self.result = 0
 142             return testcase.TestCase.EX_RUN_ERROR
 143         self.result = 100
 144         return testcase.TestCase.EX_OK
 145
 146     def clean(self):
 147         """
 148         Cleanup all OpenStack objects. Should be called on completion
 149         :return:
 150         """
 151         assert self.cloud
 152         self.cloud.delete_server(self.vm1, wait=True)
 153         self.cloud.delete_image(self.image)
 154         self.cloud.remove_router_interface(self.router, self.subnet.id)
 155         self.cloud.delete_router(self.router.id)
 156         self.cloud.delete_network(self.network.id)
 157         self.cloud.delete_flavor(self.flavor.id)
 158
 159     def _do_vping(self):
 160         """
 161         Method to be implemented by subclasses
 162         Begins the real test after the OpenStack environment has been setup
 163         :return: T/F
 164         """
 165         raise NotImplementedError('vping execution is not implemented')