"Fossies" - the Fresh Open Source Software Archive

Member "cinder-14.0.2/cinder/tests/functional/functional_helpers.py" (4 Oct 2019, 7859 Bytes) of package /linux/misc/openstack/cinder-14.0.2.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file.

    1 # Copyright 2011 Justin Santa Barbara
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 
   16 """
   17 Provides common functionality for functional tests
   18 """
   19 import os.path
   20 import random
   21 import string
   22 import time
   23 import uuid
   24 
   25 import fixtures
   26 import mock
   27 from oslo_config import cfg
   28 
   29 from cinder import service
   30 from cinder import test  # For the flags
   31 from cinder.tests.functional.api import client
   32 from cinder.tests.unit import fake_constants as fake
   33 
   34 CONF = cfg.CONF
   35 VOLUME = 'VOLUME'
   36 GROUP = 'GROUP'
   37 GROUP_SNAPSHOT = 'GROUP_SNAPSHOT'
   38 
   39 
   40 def generate_random_alphanumeric(length):
   41     """Creates a random alphanumeric string of specified length."""
   42     return ''.join(random.choice(string.ascii_uppercase + string.digits)
   43                    for _x in range(length))
   44 
   45 
   46 def generate_random_numeric(length):
   47     """Creates a random numeric string of specified length."""
   48     return ''.join(random.choice(string.digits)
   49                    for _x in range(length))
   50 
   51 
   52 def generate_new_element(items, prefix, numeric=False):
   53     """Creates a random string with prefix, that is not in 'items' list."""
   54     while True:
   55         if numeric:
   56             candidate = prefix + generate_random_numeric(8)
   57         else:
   58             candidate = prefix + generate_random_alphanumeric(8)
   59         if candidate not in items:
   60             return candidate
   61 
   62 
   63 class _FunctionalTestBase(test.TestCase):
   64     osapi_version_major = '3'
   65     osapi_version_minor = '0'
   66 
   67     def setUp(self):
   68         super(_FunctionalTestBase, self).setUp()
   69 
   70         f = self._get_flags()
   71         for k, value_dict in f.items():
   72             self.override_config(k, value_dict['v'], value_dict.get('g'))
   73 
   74         for var in ('http_proxy', 'HTTP_PROXY'):
   75             self.useFixture(fixtures.EnvironmentVariable(var))
   76 
   77         # set up services
   78         self.volume = self.start_service('volume')
   79         # NOTE(dulek): Mocking eventlet.sleep so test won't time out on
   80         # scheduler service start.
   81         with mock.patch('eventlet.sleep'):
   82             self.scheduler = self.start_service('scheduler')
   83         self._start_api_service()
   84         self.addCleanup(self.osapi.stop)
   85 
   86         api_version = self.osapi_version_major + '.' + self.osapi_version_minor
   87         self.api = client.TestOpenStackClient(fake.USER_ID,
   88                                               fake.PROJECT_ID, self.auth_url,
   89                                               api_version)
   90 
   91     def _update_project(self, new_project_id):
   92         self.api.update_project(new_project_id)
   93 
   94     def _start_api_service(self):
   95         default_conf = os.path.abspath(os.path.join(
   96             os.path.dirname(__file__), '..', '..', '..',
   97             'etc/cinder/api-paste.ini'))
   98         CONF.api_paste_config = default_conf
   99         self.osapi = service.WSGIService("osapi_volume")
  100         self.osapi.start()
  101         # FIXME(ja): this is not the auth url - this is the service url
  102         # FIXME(ja): this needs fixed in nova as well
  103         self.auth_url = 'http://%s:%s/v' % (self.osapi.host, self.osapi.port)
  104         self.auth_url += self.osapi_version_major
  105 
  106     def _get_flags(self):
  107         """An opportunity to setup flags, before the services are started."""
  108         f = {}
  109 
  110         # Ensure tests only listen on localhost
  111         f['osapi_volume_listen'] = {'v': '127.0.0.1'}
  112 
  113         # Auto-assign ports to allow concurrent tests
  114         f['osapi_volume_listen_port'] = {'v': 0}
  115 
  116         # Use simple scheduler to avoid complications - we test schedulers
  117         # separately
  118         f['scheduler_driver'] = {'v': ('cinder.scheduler.filter_scheduler.'
  119                                        'FilterScheduler')}
  120 
  121         return f
  122 
  123     def get_unused_server_name(self):
  124         servers = self.api.get_servers()
  125         server_names = [server['name'] for server in servers]
  126         return generate_new_element(server_names, 'server')
  127 
  128     def get_invalid_image(self):
  129         return str(uuid.uuid4())
  130 
  131     def _build_minimal_create_server_request(self):
  132         server = {}
  133 
  134         image = self.api.get_images()[0]
  135 
  136         if 'imageRef' in image:
  137             image_href = image['imageRef']
  138         else:
  139             image_href = image['id']
  140             image_href = 'http://fake.server/%s' % image_href
  141 
  142         # We now have a valid imageId
  143         server['imageRef'] = image_href
  144 
  145         # Set a valid flavorId
  146         flavor = self.api.get_flavors()[0]
  147         server['flavorRef'] = 'http://fake.server/%s' % flavor['id']
  148 
  149         # Set a valid server name
  150         server_name = self.get_unused_server_name()
  151         server['name'] = server_name
  152         return server
  153 
  154     def _poll_resource_while(self, res_id, continue_states, res_type=VOLUME,
  155                              expected_end_status=None, max_retries=5,
  156                              status_field='status'):
  157         """Poll (briefly) while the state is in continue_states.
  158 
  159         Continues until the state changes from continue_states or max_retries
  160         are hit. If expected_end_status is specified, we assert that the end
  161         status of the resource is expected_end_status.
  162         """
  163         retries = 0
  164         while retries <= max_retries:
  165             try:
  166                 if res_type == VOLUME:
  167                     found_res = self.api.get_volume(res_id)
  168                 elif res_type == GROUP:
  169                     found_res = self.api.get_group(res_id)
  170                 elif res_type == GROUP_SNAPSHOT:
  171                     found_res = self.api.get_group_snapshot(res_id)
  172                 else:
  173                     return None
  174             except client.OpenStackApiException404:
  175                 return None
  176             except client.OpenStackApiException:
  177                 # NOTE(xyang): Got OpenStackApiException(
  178                 # u'Unexpected status code',) sometimes, but
  179                 # it works if continue.
  180                 continue
  181 
  182             self.assertEqual(res_id, found_res['id'])
  183             res_status = found_res[status_field]
  184             if res_status not in continue_states:
  185                 if expected_end_status:
  186                     self.assertEqual(expected_end_status, res_status)
  187                 return found_res
  188 
  189             time.sleep(1)
  190             retries += 1
  191 
  192     def _poll_volume_while(self, volume_id, continue_states,
  193                            expected_end_status=None, max_retries=5,
  194                            status_field='status'):
  195         return self._poll_resource_while(volume_id, continue_states,
  196                                          VOLUME, expected_end_status,
  197                                          max_retries, status_field)
  198 
  199     def _poll_group_while(self, group_id, continue_states,
  200                           expected_end_status=None, max_retries=30,
  201                           status_field='status'):
  202         return self._poll_resource_while(group_id, continue_states,
  203                                          GROUP, expected_end_status,
  204                                          max_retries, status_field)
  205 
  206     def _poll_group_snapshot_while(self, group_snapshot_id, continue_states,
  207                                    expected_end_status=None, max_retries=30):
  208         return self._poll_resource_while(group_snapshot_id, continue_states,
  209                                          GROUP_SNAPSHOT, expected_end_status,
  210                                          max_retries)