functest/opnfv_tests/openstack/cinder/cinder_base.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2018 Enea AB 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 cinder_test"""
  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 from functest.utils import functest_utils
  24
  25
  26 class CinderBase(testcase.TestCase):
  27
  28     """
  29     Base class for CinderCheck tests that check volume data persistence
  30     between two VMs shared internal network.
  31     This class is responsible for creating the image, internal network,
  32     flavor, volume.
  33     """
  34     # pylint: disable=too-many-instance-attributes
  35
  36     def __init__(self, **kwargs):
  37         super(CinderBase, self).__init__(**kwargs)
  38         self.logger = logging.getLogger(__name__)
  39         self.cloud = os_client_config.make_shade()
  40         self.ext_net = functest_utils.get_external_network(self.cloud)
  41         self.logger.debug("ext_net: %s", self.ext_net)
  42         self.guid = '-' + str(uuid.uuid4())
  43         self.network = None
  44         self.subnet = None
  45         self.router = None
  46         self.image = None
  47         self.flavor = None
  48         self.vm1 = None
  49         self.volume = None
  50
  51     def run(self, **kwargs):  # pylint: disable=too-many-locals
  52         """
  53         Begins the test execution which should originate from the subclass
  54         """
  55         assert self.cloud
  56         assert self.ext_net
  57         self.logger.info('Begin virtual environment setup')
  58
  59         self.start_time = time.time()
  60         self.logger.info(
  61             "CinderCheck Start Time:'%s'",
  62             datetime.fromtimestamp(self.start_time).strftime(
  63                 '%Y-%m-%d %H:%M:%S'))
  64
  65         image_base_name = '{}-{}'.format(
  66             getattr(config.CONF, 'cinder_image_name'), self.guid)
  67         self.logger.info("Creating image with name: '%s'", image_base_name)
  68         meta = getattr(config.CONF, 'openstack_extra_properties', None)
  69         self.logger.info("Image metadata: %s", meta)
  70         self.image = self.cloud.create_image(
  71             image_base_name,
  72             filename=getattr(config.CONF, 'openstack_image_url'),
  73             meta=meta)
  74         self.logger.debug("image: %s", self.image)
  75
  76         private_net_name = getattr(
  77             config.CONF, 'cinder_private_net_name') + self.guid
  78         private_subnet_name = str(getattr(
  79             config.CONF, 'cinder_private_subnet_name') + self.guid)
  80         private_subnet_cidr = getattr(
  81             config.CONF, 'cinder_private_subnet_cidr')
  82
  83         provider = {}
  84         if hasattr(config.CONF, 'cinder_network_type'):
  85             provider["network_type"] = getattr(
  86                 config.CONF, 'cinder_network_type')
  87         if hasattr(config.CONF, 'cinder_physical_network'):
  88             provider["physical_network"] = getattr(
  89                 config.CONF, 'cinder_physical_network')
  90         if hasattr(config.CONF, 'cinder_segmentation_id'):
  91             provider["segmentation_id"] = getattr(
  92                 config.CONF, 'cinder_segmentation_id')
  93         self.logger.info(
  94             "Creating network with name: '%s'", private_net_name)
  95         self.network = self.cloud.create_network(
  96             private_net_name,
  97             provider=provider)
  98         self.logger.debug("network: %s", self.network)
  99
 100         self.subnet = self.cloud.create_subnet(
 101             self.network.id,
 102             subnet_name=private_subnet_name,
 103             cidr=private_subnet_cidr,
 104             enable_dhcp=True,
 105             dns_nameservers=[env.get('NAMESERVER')])
 106         self.logger.debug("subnet: %s", self.subnet)
 107
 108         router_name = getattr(config.CONF, 'cinder_router_name') + self.guid
 109         self.logger.info("Creating router with name: '%s'", router_name)
 110         self.router = self.cloud.create_router(
 111             name=router_name,
 112             ext_gateway_net_id=self.ext_net.id)
 113         self.logger.debug("router: %s", self.router)
 114         self.cloud.add_router_interface(self.router, subnet_id=self.subnet.id)
 115
 116         flavor_name = 'cinder-flavor' + self.guid
 117         self.logger.info(
 118             "Creating flavor with name: '%s'", flavor_name)
 119         self.flavor = self.cloud.create_flavor(
 120             flavor_name, getattr(config.CONF, 'openstack_flavor_ram'),
 121             getattr(config.CONF, 'openstack_flavor_vcpus'),
 122             getattr(config.CONF, 'openstack_flavor_disk'))
 123         self.logger.debug("flavor: %s", self.flavor)
 124         self.cloud.set_flavor_specs(
 125             self.flavor.id, getattr(config.CONF, 'flavor_extra_specs', {}))
 126         volume_name = 'cinder-volume' + self.guid
 127         self.logger.info(
 128             "Creating volume with name: %s", volume_name)
 129         self.volume = self.cloud.create_volume(
 130             name=volume_name, size='2')
 131         self.logger.info("volume: %s", self.volume)
 132
 133     def execute(self):
 134         """
 135         Method called by subclasses after environment has been setup
 136         :return: the exit code
 137         """
 138         self.logger.info('Begin test execution')
 139         self.stop_time = time.time()
 140         write_data = self.write_data()
 141         if write_data == testcase.TestCase.EX_OK:
 142             result = self.read_data()
 143         else:
 144             result = testcase.TestCase.EX_RUN_ERROR
 145         if result != testcase.TestCase.EX_OK:
 146             self.result = 0
 147             return testcase.TestCase.EX_RUN_ERROR
 148         self.result = 100
 149         return testcase.TestCase.EX_OK
 150
 151     def clean(self):
 152         """
 153         Cleanup all OpenStack objects. Should be called on completion
 154         :return:
 155         """
 156         assert self.cloud
 157         self.cloud.delete_image(self.image)
 158         self.cloud.remove_router_interface(self.router, self.subnet.id)
 159         self.cloud.delete_router(self.router.id)
 160         self.cloud.delete_network(self.network.id)
 161         self.cloud.delete_flavor(self.flavor.id)
 162         self.cloud.delete_volume(self.volume.id)
 163
 164     def write_data(self):
 165         """
 166         Method to be implemented by subclasses
 167         Begins the real test after the OpenStack environment has been setup
 168         :return: T/F
 169         """
 170         raise NotImplementedError('cinder test execution is not implemented')
 171
 172     def read_data(self):
 173         """
 174         Method to be implemented by subclasses
 175         Begins the real test after the OpenStack environment has been setup
 176         :return: T/F
 177         """
 178         raise NotImplementedError('cinder test execution is not implemented')