"Fossies" - the Fresh Open Source Software Archive

Member "masakari-9.0.0/masakari/tests/unit/api/openstack/test_common.py" (13 May 2020, 14458 Bytes) of package /linux/misc/openstack/masakari-9.0.0.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. See also the latest Fossies "Diffs" side-by-side code changes report for "test_common.py": 8.0.0_vs_9.0.0.

    1 # Copyright 2016 NTT DATA
    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 Test suites for 'common' code used throughout the OpenStack HTTP API.
   18 """
   19 
   20 from testtools import matchers
   21 from unittest import mock
   22 
   23 import webob
   24 
   25 from masakari.api.openstack import common
   26 from masakari import test
   27 from masakari.tests.unit.api.openstack import fakes
   28 from masakari.tests import uuidsentinel
   29 
   30 
   31 class MiscFunctionsTest(test.TestCase):
   32 
   33     def test_remove_trailing_version_from_href(self):
   34         fixture = 'http://www.testsite.com/v1.1'
   35         expected = 'http://www.testsite.com'
   36         actual = common.remove_trailing_version_from_href(fixture)
   37         self.assertEqual(actual, expected)
   38 
   39     def test_remove_trailing_version_from_href_2(self):
   40         fixture = 'http://www.testsite.com/ha/v1.1'
   41         expected = 'http://www.testsite.com/ha'
   42         actual = common.remove_trailing_version_from_href(fixture)
   43         self.assertEqual(actual, expected)
   44 
   45     def test_remove_trailing_version_from_href_3(self):
   46         fixture = 'http://www.testsite.com/v1.1/images/v10.5'
   47         expected = 'http://www.testsite.com/v1.1/images'
   48         actual = common.remove_trailing_version_from_href(fixture)
   49         self.assertEqual(actual, expected)
   50 
   51     def test_remove_trailing_version_from_href_bad_request(self):
   52         fixture = 'http://www.testsite.com/v1.1/images'
   53         self.assertRaises(ValueError,
   54                           common.remove_trailing_version_from_href,
   55                           fixture)
   56 
   57     def test_remove_trailing_version_from_href_bad_request_2(self):
   58         fixture = 'http://www.testsite.com/images/v'
   59         self.assertRaises(ValueError,
   60                           common.remove_trailing_version_from_href,
   61                           fixture)
   62 
   63     def test_remove_trailing_version_from_href_bad_request_3(self):
   64         fixture = 'http://www.testsite.com/v1.1images'
   65         self.assertRaises(ValueError,
   66                           common.remove_trailing_version_from_href,
   67                           fixture)
   68 
   69 
   70 class TestCollectionLinks(test.NoDBTestCase):
   71     """Tests the _get_collection_links method."""
   72 
   73     @mock.patch('masakari.api.openstack.common.ViewBuilder._get_next_link')
   74     def test_items_less_than_limit(self, href_link_mock):
   75         items = [
   76             {"uuid": "123"}
   77         ]
   78         req = mock.MagicMock()
   79         params = mock.PropertyMock(return_value=dict(limit=10))
   80         type(req).params = params
   81 
   82         builder = common.ViewBuilder()
   83         results = builder._get_collection_links(req, items, "ignored", "uuid")
   84 
   85         self.assertFalse(href_link_mock.called)
   86         self.assertThat(results, matchers.HasLength(0))
   87 
   88     @mock.patch('masakari.api.openstack.common.ViewBuilder._get_next_link')
   89     def test_items_equals_given_limit(self, href_link_mock):
   90         items = [
   91             {"uuid": "123"}
   92         ]
   93         req = mock.MagicMock()
   94         params = mock.PropertyMock(return_value=dict(limit=1))
   95         type(req).params = params
   96 
   97         builder = common.ViewBuilder()
   98         results = builder._get_collection_links(req, items,
   99                                                 mock.sentinel.coll_key,
  100                                                 "uuid")
  101 
  102         href_link_mock.assert_called_once_with(req, "123",
  103                                                mock.sentinel.coll_key)
  104         self.assertThat(results, matchers.HasLength(1))
  105 
  106     @mock.patch('masakari.api.openstack.common.ViewBuilder._get_next_link')
  107     def test_items_equals_default_limit(self, href_link_mock):
  108         items = [
  109             {"uuid": "123"}
  110         ]
  111         req = mock.MagicMock()
  112         params = mock.PropertyMock(return_value=dict())
  113         type(req).params = params
  114         self.flags(osapi_max_limit=1)
  115 
  116         builder = common.ViewBuilder()
  117         results = builder._get_collection_links(req, items,
  118                                                 mock.sentinel.coll_key,
  119                                                 "uuid")
  120 
  121         href_link_mock.assert_called_once_with(req, "123",
  122                                                mock.sentinel.coll_key)
  123         self.assertThat(results, matchers.HasLength(1))
  124 
  125     @mock.patch('masakari.api.openstack.common.ViewBuilder._get_next_link')
  126     def test_items_equals_default_limit_with_given(self, href_link_mock):
  127         items = [
  128             {"uuid": "123"}
  129         ]
  130         req = mock.MagicMock()
  131         # Given limit is greater than default max, only return default max
  132         params = mock.PropertyMock(return_value=dict(limit=2))
  133         type(req).params = params
  134         self.flags(osapi_max_limit=1)
  135 
  136         builder = common.ViewBuilder()
  137         results = builder._get_collection_links(req, items,
  138                                                 mock.sentinel.coll_key,
  139                                                 "uuid")
  140 
  141         href_link_mock.assert_called_once_with(req, "123",
  142                                                mock.sentinel.coll_key)
  143         self.assertThat(results, matchers.HasLength(1))
  144 
  145 
  146 class LinkPrefixTest(test.NoDBTestCase):
  147 
  148     def test_update_link_prefix(self):
  149         vb = common.ViewBuilder()
  150         result = vb._update_link_prefix("http://192.168.0.243:24/",
  151                                         "http://127.0.0.1/ha")
  152         self.assertEqual("http://127.0.0.1/ha", result)
  153 
  154         result = vb._update_link_prefix("http://foo.x.com/v1",
  155                                         "http://new.prefix.com")
  156         self.assertEqual("http://new.prefix.com/v1", result)
  157 
  158         result = vb._update_link_prefix("http://foo.x.com/v1",
  159                                         "http://new.prefix.com:20455/"
  160                                         "new_extra_prefix")
  161         self.assertEqual("http://new.prefix.com:20455/new_extra_prefix/v1",
  162                          result)
  163 
  164 
  165 class UrlJoinTest(test.NoDBTestCase):
  166     def test_url_join(self):
  167         pieces = ["one", "two", "three"]
  168         joined = common.url_join(*pieces)
  169         self.assertEqual("one/two/three", joined)
  170 
  171     def test_url_join_extra_slashes(self):
  172         pieces = ["one/", "/two//", "/three/"]
  173         joined = common.url_join(*pieces)
  174         self.assertEqual("one/two/three", joined)
  175 
  176     def test_url_join_trailing_slash(self):
  177         pieces = ["one", "two", "three", ""]
  178         joined = common.url_join(*pieces)
  179         self.assertEqual("one/two/three/", joined)
  180 
  181     def test_url_join_empty_list(self):
  182         pieces = []
  183         joined = common.url_join(*pieces)
  184         self.assertEqual("", joined)
  185 
  186     def test_url_join_single_empty_string(self):
  187         pieces = [""]
  188         joined = common.url_join(*pieces)
  189         self.assertEqual("", joined)
  190 
  191     def test_url_join_single_slash(self):
  192         pieces = ["/"]
  193         joined = common.url_join(*pieces)
  194         self.assertEqual("", joined)
  195 
  196 
  197 class ViewBuilderLinkTest(test.NoDBTestCase):
  198     project_id = uuidsentinel.fake_project_id
  199     api_version = "1.0"
  200 
  201     def setUp(self):
  202         super(ViewBuilderLinkTest, self).setUp()
  203         self.request = self.req("/%s" % self.project_id)
  204         self.vb = common.ViewBuilder()
  205 
  206     def req(self, url, use_admin_context=False):
  207         return fakes.HTTPRequest.blank(url,
  208                 use_admin_context=use_admin_context, version=self.api_version)
  209 
  210     def test_get_project_id(self):
  211         proj_id = self.vb._get_project_id(self.request)
  212         self.assertEqual(self.project_id, proj_id)
  213 
  214     def test_get_next_link(self):
  215         identifier = "identifier"
  216         collection = "collection"
  217         next_link = self.vb._get_next_link(self.request, identifier,
  218                                            collection)
  219         expected = "/".join((self.request.url,
  220                              "%s?marker=%s" % (collection, identifier)))
  221         self.assertEqual(expected, next_link)
  222 
  223     def test_get_href_link(self):
  224         identifier = "identifier"
  225         collection = "collection"
  226         href_link = self.vb._get_href_link(self.request, identifier,
  227                                            collection)
  228         expected = "/".join((self.request.url, collection, identifier))
  229         self.assertEqual(expected, href_link)
  230 
  231     def test_get_bookmark_link(self):
  232         identifier = "identifier"
  233         collection = "collection"
  234         bookmark_link = self.vb._get_bookmark_link(self.request, identifier,
  235                                                    collection)
  236         bmk_url = (
  237             common.remove_trailing_version_from_href((
  238                 self.request.application_url)))
  239         expected = "/".join((bmk_url, self.project_id, collection, identifier))
  240         self.assertEqual(expected, bookmark_link)
  241 
  242 
  243 class PaginationParamsTest(test.NoDBTestCase):
  244     """Unit tests for the `masakari.api.openstack.common.get_pagination_params`
  245     method which takes in a request object and returns 'marker' and 'limit'
  246     GET params.
  247     """
  248 
  249     def test_no_params(self):
  250         # Test no params.
  251         req = webob.Request.blank('/')
  252         self.assertEqual(common.get_pagination_params(req), {})
  253 
  254     def test_valid_marker(self):
  255         # Test valid marker param.
  256         req = webob.Request.blank('/?marker=263abb28-1de6-412f-b00'
  257                                   'b-f0ee0c4333c2')
  258         self.assertEqual(common.get_pagination_params(req),
  259                          {'marker': '263abb28-1de6-412f-b00b-f0ee0c4333c2'})
  260 
  261     def test_valid_limit(self):
  262         # Test valid limit param.
  263         req = webob.Request.blank('/?limit=10')
  264         self.assertEqual(common.get_pagination_params(req), {'limit': 10})
  265 
  266     def test_invalid_limit(self):
  267         # Test invalid limit param.
  268         req = webob.Request.blank('/?limit=-2')
  269         self.assertRaises(
  270             webob.exc.HTTPBadRequest, common.get_pagination_params, req)
  271 
  272     def test_valid_limit_and_marker(self):
  273         # Test valid limit and marker parameters.
  274         marker = '263abb28-1de6-412f-b00b-f0ee0c4333c2'
  275         req = webob.Request.blank('/?limit=20&marker=%s' % marker)
  276         self.assertEqual(common.get_pagination_params(req),
  277                          {'marker': marker, 'limit': 20})
  278 
  279     def test_valid_page_size(self):
  280         # Test valid page_size param.
  281         req = webob.Request.blank('/?page_size=10')
  282         self.assertEqual(common.get_pagination_params(req),
  283                          {'page_size': 10})
  284 
  285     def test_invalid_page_size(self):
  286         # Test invalid page_size param.
  287         req = webob.Request.blank('/?page_size=-2')
  288         self.assertRaises(
  289             webob.exc.HTTPBadRequest, common.get_pagination_params, req)
  290 
  291     def test_valid_limit_and_page_size(self):
  292         # Test valid limit and page_size parameters.
  293         req = webob.Request.blank('/?limit=20&page_size=5')
  294         self.assertEqual(common.get_pagination_params(req),
  295                          {'page_size': 5, 'limit': 20})
  296 
  297 
  298 class SortParamTest(test.NoDBTestCase):
  299 
  300     def test_get_sort_params_defaults(self):
  301         # Verifies the default sort key and direction.
  302         sort_keys, sort_dirs = common.get_sort_params({})
  303         self.assertEqual(['created_at'], sort_keys)
  304         self.assertEqual(['desc'], sort_dirs)
  305 
  306     def test_get_sort_params_override_defaults(self):
  307         # Verifies that the defaults can be overriden.
  308         sort_keys, sort_dirs = common.get_sort_params({}, default_key='key1',
  309                                                       default_dir='dir1')
  310         self.assertEqual(['key1'], sort_keys)
  311         self.assertEqual(['dir1'], sort_dirs)
  312 
  313         sort_keys, sort_dirs = common.get_sort_params({}, default_key=None,
  314                                                       default_dir=None)
  315         self.assertEqual([], sort_keys)
  316         self.assertEqual([], sort_dirs)
  317 
  318     def test_get_sort_params_single_value(self):
  319         # Verifies a single sort key and direction.
  320         params = webob.multidict.MultiDict()
  321         params.add('sort_key', 'key1')
  322         params.add('sort_dir', 'dir1')
  323         sort_keys, sort_dirs = common.get_sort_params(params)
  324         self.assertEqual(['key1'], sort_keys)
  325         self.assertEqual(['dir1'], sort_dirs)
  326 
  327     def test_get_sort_params_single_with_default(self):
  328         # Verifies a single sort value with a default.
  329         params = webob.multidict.MultiDict()
  330         params.add('sort_key', 'key1')
  331         sort_keys, sort_dirs = common.get_sort_params(params)
  332         self.assertEqual(['key1'], sort_keys)
  333         # sort_key was supplied, sort_dir should be defaulted
  334         self.assertEqual(['desc'], sort_dirs)
  335 
  336         params = webob.multidict.MultiDict()
  337         params.add('sort_dir', 'dir1')
  338         sort_keys, sort_dirs = common.get_sort_params(params)
  339         self.assertEqual(['created_at'], sort_keys)
  340         # sort_dir was supplied, sort_key should be defaulted
  341         self.assertEqual(['dir1'], sort_dirs)
  342 
  343     def test_get_sort_params_multiple_values(self):
  344         # Verifies multiple sort parameter values.
  345         params = webob.multidict.MultiDict()
  346         params.add('sort_key', 'key1')
  347         params.add('sort_key', 'key2')
  348         params.add('sort_key', 'key3')
  349         params.add('sort_dir', 'dir1')
  350         params.add('sort_dir', 'dir2')
  351         params.add('sort_dir', 'dir3')
  352         sort_keys, sort_dirs = common.get_sort_params(params)
  353         self.assertEqual(['key1', 'key2', 'key3'], sort_keys)
  354         self.assertEqual(['dir1', 'dir2', 'dir3'], sort_dirs)
  355         # Also ensure that the input parameters are not modified
  356         sort_key_vals = []
  357         sort_dir_vals = []
  358         while 'sort_key' in params:
  359             sort_key_vals.append(params.pop('sort_key'))
  360         while 'sort_dir' in params:
  361             sort_dir_vals.append(params.pop('sort_dir'))
  362         self.assertEqual(['key1', 'key2', 'key3'], sort_key_vals)
  363         self.assertEqual(['dir1', 'dir2', 'dir3'], sort_dir_vals)
  364         self.assertEqual(0, len(params))