"Fossies" - the Fresh Open Source Software Archive

Member "glance-20.0.1/glance/tests/unit/common/test_utils.py" (12 Aug 2020, 30237 Bytes) of package /linux/misc/openstack/glance-20.0.1.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_utils.py": 20.0.0_vs_20.0.1.

    1 # Copyright 2011 OpenStack Foundation
    2 # Copyright 2015 Mirantis, Inc
    3 # All Rights Reserved.
    4 #
    5 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    6 #    not use this file except in compliance with the License. You may obtain
    7 #    a copy of the License at
    8 #
    9 #         http://www.apache.org/licenses/LICENSE-2.0
   10 #
   11 #    Unless required by applicable law or agreed to in writing, software
   12 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   13 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   14 #    License for the specific language governing permissions and limitations
   15 #    under the License.
   16 
   17 import glance_store as store
   18 import tempfile
   19 from unittest import mock
   20 
   21 from oslo_config import cfg
   22 from oslo_log import log as logging
   23 import six
   24 import webob
   25 
   26 from glance.common import exception
   27 from glance.common import store_utils
   28 from glance.common import utils
   29 from glance.tests import utils as test_utils
   30 
   31 
   32 CONF = cfg.CONF
   33 
   34 
   35 class TestStoreUtils(test_utils.BaseTestCase):
   36     """Test glance.common.store_utils module"""
   37 
   38     def _test_update_store_in_location(self, metadata, store_id, expected,
   39                                        store_id_call_count=1,
   40                                        save_call_count=1):
   41         image = mock.Mock()
   42         image_repo = mock.Mock()
   43         image_repo.save = mock.Mock()
   44         locations = [{
   45             'url': 'rbd://aaaaaaaa/images/id',
   46             'metadata': metadata
   47         }]
   48         image.locations = locations
   49         with mock.patch.object(
   50                 store_utils, '_get_store_id_from_uri') as mock_get_store_id:
   51             mock_get_store_id.return_value = store_id
   52             store_utils.update_store_in_locations(image, image_repo)
   53             self.assertEqual(image.locations[0]['metadata'].get(
   54                 'store'), expected)
   55             self.assertEqual(store_id_call_count, mock_get_store_id.call_count)
   56             self.assertEqual(save_call_count, image_repo.save.call_count)
   57 
   58     def test_update_store_location_with_no_store(self):
   59         enabled_backends = {
   60             "rbd1": "rbd",
   61             "rbd2": "rbd"
   62         }
   63         self.config(enabled_backends=enabled_backends)
   64         self._test_update_store_in_location({}, 'rbd1', 'rbd1')
   65 
   66     def test_update_store_location_with_different_store(self):
   67         enabled_backends = {
   68             "ceph1": "rbd",
   69             "ceph2": "rbd"
   70         }
   71         self.config(enabled_backends=enabled_backends)
   72         self._test_update_store_in_location(
   73             {'store': 'rbd2'}, 'ceph1', 'ceph1')
   74 
   75     def test_update_store_location_with_same_store(self):
   76         enabled_backends = {
   77             "rbd1": "rbd",
   78             "rbd2": "rbd"
   79         }
   80         self.config(enabled_backends=enabled_backends)
   81         self._test_update_store_in_location({'store': 'rbd1'}, 'rbd1', 'rbd1',
   82                                             store_id_call_count=0,
   83                                             save_call_count=0)
   84 
   85     def test_update_store_location_with_store_none(self):
   86         enabled_backends = {
   87             "rbd1": "rbd",
   88             "rbd2": "rbd"
   89         }
   90         self.config(enabled_backends=enabled_backends)
   91         self._test_update_store_in_location({}, None, None,
   92                                             save_call_count=0)
   93 
   94 
   95 class TestUtils(test_utils.BaseTestCase):
   96     """Test routines in glance.utils"""
   97 
   98     def test_cooperative_reader(self):
   99         """Ensure cooperative reader class accesses all bytes of file"""
  100         BYTES = 1024
  101         bytes_read = 0
  102         with tempfile.TemporaryFile('w+') as tmp_fd:
  103             tmp_fd.write('*' * BYTES)
  104             tmp_fd.seek(0)
  105             for chunk in utils.CooperativeReader(tmp_fd):
  106                 bytes_read += len(chunk)
  107 
  108         self.assertEqual(BYTES, bytes_read)
  109 
  110         bytes_read = 0
  111         with tempfile.TemporaryFile('w+') as tmp_fd:
  112             tmp_fd.write('*' * BYTES)
  113             tmp_fd.seek(0)
  114             reader = utils.CooperativeReader(tmp_fd)
  115             byte = reader.read(1)
  116             while len(byte) != 0:
  117                 bytes_read += 1
  118                 byte = reader.read(1)
  119 
  120         self.assertEqual(BYTES, bytes_read)
  121 
  122     def test_cooperative_reader_of_iterator(self):
  123         """Ensure cooperative reader supports iterator backends too"""
  124         data = b'abcdefgh'
  125         data_list = [data[i:i + 1] * 3 for i in range(len(data))]
  126         reader = utils.CooperativeReader(data_list)
  127         chunks = []
  128         while True:
  129             chunks.append(reader.read(3))
  130             if chunks[-1] == b'':
  131                 break
  132         meat = b''.join(chunks)
  133         self.assertEqual(b'aaabbbcccdddeeefffggghhh', meat)
  134 
  135     def test_cooperative_reader_unbounded_read_on_iterator(self):
  136         """Ensure cooperative reader is happy with empty iterators"""
  137         data = b'abcdefgh'
  138         data_list = [data[i:i + 1] * 3 for i in range(len(data))]
  139         reader = utils.CooperativeReader(data_list)
  140         self.assertEqual(
  141             [chunk for chunk in iter(lambda: reader.read(), b'')],
  142             [b'aaa', b'bbb', b'ccc', b'ddd', b'eee', b'fff', b'ggg', b'hhh'])
  143 
  144     def test_cooperative_reader_on_iterator_with_buffer(self):
  145         """Ensure cooperative reader is happy with empty iterators"""
  146         data_list = [b'abcd', b'efgh']
  147         reader = utils.CooperativeReader(data_list)
  148         # read from part of a chunk, get the first item into the buffer
  149         self.assertEqual(b'ab', reader.read(2))
  150         # read purely from buffer
  151         self.assertEqual(b'c', reader.read(1))
  152         # unbounded read grabs the rest of the buffer
  153         self.assertEqual(b'd', reader.read())
  154         # then the whole next chunk
  155         self.assertEqual(b'efgh', reader.read())
  156         # past that, it's always empty
  157         self.assertEqual(b'', reader.read())
  158 
  159     def test_cooperative_reader_unbounded_read_on_empty_iterator(self):
  160         """Ensure cooperative reader is happy with empty iterators"""
  161         reader = utils.CooperativeReader([])
  162         self.assertEqual(b'', reader.read())
  163 
  164     def test_cooperative_reader_of_iterator_stop_iteration_err(self):
  165         """Ensure cooperative reader supports iterator backends too"""
  166         reader = utils.CooperativeReader([l * 3 for l in ''])
  167         chunks = []
  168         while True:
  169             chunks.append(reader.read(3))
  170             if chunks[-1] == b'':
  171                 break
  172         meat = b''.join(chunks)
  173         self.assertEqual(b'', meat)
  174 
  175     def _create_generator(self, chunk_size, max_iterations):
  176         chars = b'abc'
  177         iteration = 0
  178         while True:
  179             index = iteration % len(chars)
  180             chunk = chars[index:index + 1] * chunk_size
  181             yield chunk
  182             iteration += 1
  183             if iteration >= max_iterations:
  184                 return
  185 
  186     def _test_reader_chunked(self, chunk_size, read_size, max_iterations=5):
  187         generator = self._create_generator(chunk_size, max_iterations)
  188         reader = utils.CooperativeReader(generator)
  189         result = bytearray()
  190         while True:
  191             data = reader.read(read_size)
  192             if len(data) == 0:
  193                 break
  194             self.assertLessEqual(len(data), read_size)
  195             result += data
  196         expected = (b'a' * chunk_size +
  197                     b'b' * chunk_size +
  198                     b'c' * chunk_size +
  199                     b'a' * chunk_size +
  200                     b'b' * chunk_size)
  201         self.assertEqual(expected, bytes(result))
  202 
  203     def test_cooperative_reader_preserves_size_chunk_less_then_read(self):
  204         self._test_reader_chunked(43, 101)
  205 
  206     def test_cooperative_reader_preserves_size_chunk_equals_read(self):
  207         self._test_reader_chunked(1024, 1024)
  208 
  209     def test_cooperative_reader_preserves_size_chunk_more_then_read(self):
  210         chunk_size = 16 * 1024 * 1024  # 16 Mb, as in remote http source
  211         read_size = 8 * 1024           # 8k, as in httplib
  212         self._test_reader_chunked(chunk_size, read_size)
  213 
  214     def test_limiting_reader(self):
  215         """Ensure limiting reader class accesses all bytes of file"""
  216         BYTES = 1024
  217         bytes_read = 0
  218         data = six.StringIO("*" * BYTES)
  219         for chunk in utils.LimitingReader(data, BYTES):
  220             bytes_read += len(chunk)
  221 
  222         self.assertEqual(BYTES, bytes_read)
  223 
  224         bytes_read = 0
  225         data = six.StringIO("*" * BYTES)
  226         reader = utils.LimitingReader(data, BYTES)
  227         byte = reader.read(1)
  228         while len(byte) != 0:
  229             bytes_read += 1
  230             byte = reader.read(1)
  231 
  232         self.assertEqual(BYTES, bytes_read)
  233 
  234     def test_limiting_reader_fails(self):
  235         """Ensure limiting reader class throws exceptions if limit exceeded"""
  236         BYTES = 1024
  237 
  238         def _consume_all_iter():
  239             bytes_read = 0
  240             data = six.StringIO("*" * BYTES)
  241             for chunk in utils.LimitingReader(data, BYTES - 1):
  242                 bytes_read += len(chunk)
  243 
  244         self.assertRaises(exception.ImageSizeLimitExceeded, _consume_all_iter)
  245 
  246         def _consume_all_read():
  247             bytes_read = 0
  248             data = six.StringIO("*" * BYTES)
  249             reader = utils.LimitingReader(data, BYTES - 1)
  250             byte = reader.read(1)
  251             while len(byte) != 0:
  252                 bytes_read += 1
  253                 byte = reader.read(1)
  254 
  255         self.assertRaises(exception.ImageSizeLimitExceeded, _consume_all_read)
  256 
  257     def test_get_meta_from_headers(self):
  258         resp = webob.Response()
  259         resp.headers = {"x-image-meta-name": 'test',
  260                         'x-image-meta-virtual-size': 80}
  261         result = utils.get_image_meta_from_headers(resp)
  262         self.assertEqual({'name': 'test', 'properties': {},
  263                           'virtual_size': 80}, result)
  264 
  265     def test_get_meta_from_headers_none_virtual_size(self):
  266         resp = webob.Response()
  267         resp.headers = {"x-image-meta-name": 'test',
  268                         'x-image-meta-virtual-size': 'None'}
  269         result = utils.get_image_meta_from_headers(resp)
  270         self.assertEqual({'name': 'test', 'properties': {},
  271                           'virtual_size': None}, result)
  272 
  273     def test_get_meta_from_headers_bad_headers(self):
  274         resp = webob.Response()
  275         resp.headers = {"x-image-meta-bad": 'test'}
  276         self.assertRaises(webob.exc.HTTPBadRequest,
  277                           utils.get_image_meta_from_headers, resp)
  278         resp.headers = {"x-image-meta-": 'test'}
  279         self.assertRaises(webob.exc.HTTPBadRequest,
  280                           utils.get_image_meta_from_headers, resp)
  281         resp.headers = {"x-image-meta-*": 'test'}
  282         self.assertRaises(webob.exc.HTTPBadRequest,
  283                           utils.get_image_meta_from_headers, resp)
  284 
  285     def test_image_meta(self):
  286         image_meta = {'x-image-meta-size': 'test'}
  287         image_meta_properties = {'properties': {'test': "test"}}
  288         actual = utils.image_meta_to_http_headers(image_meta)
  289         actual_test2 = utils.image_meta_to_http_headers(
  290             image_meta_properties)
  291         self.assertEqual({'x-image-meta-x-image-meta-size': u'test'}, actual)
  292         self.assertEqual({'x-image-meta-property-test': u'test'},
  293                          actual_test2)
  294 
  295     def test_create_mashup_dict_with_different_core_custom_properties(self):
  296         image_meta = {
  297             'id': 'test-123',
  298             'name': 'fake_image',
  299             'status': 'active',
  300             'created_at': '',
  301             'min_disk': '10G',
  302             'min_ram': '1024M',
  303             'protected': False,
  304             'locations': '',
  305             'checksum': 'c1234',
  306             'owner': '',
  307             'disk_format': 'raw',
  308             'container_format': 'bare',
  309             'size': '123456789',
  310             'virtual_size': '123456789',
  311             'is_public': 'public',
  312             'deleted': True,
  313             'updated_at': '',
  314             'properties': {'test_key': 'test_1234'},
  315         }
  316 
  317         mashup_dict = utils.create_mashup_dict(image_meta)
  318         self.assertNotIn('properties', mashup_dict)
  319         self.assertEqual(image_meta['properties']['test_key'],
  320                          mashup_dict['test_key'])
  321 
  322     def test_create_mashup_dict_with_same_core_custom_properties(self):
  323         image_meta = {
  324             'id': 'test-123',
  325             'name': 'fake_image',
  326             'status': 'active',
  327             'created_at': '',
  328             'min_disk': '10G',
  329             'min_ram': '1024M',
  330             'protected': False,
  331             'locations': '',
  332             'checksum': 'c1234',
  333             'owner': '',
  334             'disk_format': 'raw',
  335             'container_format': 'bare',
  336             'size': '123456789',
  337             'virtual_size': '123456789',
  338             'is_public': 'public',
  339             'deleted': True,
  340             'updated_at': '',
  341             'properties': {'min_ram': '2048M'},
  342         }
  343 
  344         mashup_dict = utils.create_mashup_dict(image_meta)
  345         self.assertNotIn('properties', mashup_dict)
  346         self.assertNotEqual(image_meta['properties']['min_ram'],
  347                             mashup_dict['min_ram'])
  348         self.assertEqual(image_meta['min_ram'], mashup_dict['min_ram'])
  349 
  350     def test_mutating(self):
  351         class FakeContext(object):
  352             def __init__(self):
  353                 self.read_only = False
  354 
  355         class Fake(object):
  356             def __init__(self):
  357                 self.context = FakeContext()
  358 
  359         def fake_function(req, context):
  360             return 'test passed'
  361 
  362         req = webob.Request.blank('/some_request')
  363         result = utils.mutating(fake_function)
  364         self.assertEqual("test passed", result(req, Fake()))
  365 
  366     def test_valid_hostname(self):
  367         valid_inputs = ['localhost',
  368                         'glance04-a'
  369                         'G',
  370                         '528491']
  371 
  372         for input_str in valid_inputs:
  373             self.assertTrue(utils.is_valid_hostname(input_str))
  374 
  375     def test_valid_hostname_fail(self):
  376         invalid_inputs = ['localhost.localdomain',
  377                           '192.168.0.1',
  378                           u'\u2603',
  379                           'glance02.stack42.local']
  380 
  381         for input_str in invalid_inputs:
  382             self.assertFalse(utils.is_valid_hostname(input_str))
  383 
  384     def test_valid_fqdn(self):
  385         valid_inputs = ['localhost.localdomain',
  386                         'glance02.stack42.local'
  387                         'glance04-a.stack47.local',
  388                         'img83.glance.xn--penstack-r74e.org']
  389 
  390         for input_str in valid_inputs:
  391             self.assertTrue(utils.is_valid_fqdn(input_str))
  392 
  393     def test_valid_fqdn_fail(self):
  394         invalid_inputs = ['localhost',
  395                           '192.168.0.1',
  396                           '999.88.77.6',
  397                           u'\u2603.local',
  398                           'glance02.stack42']
  399 
  400         for input_str in invalid_inputs:
  401             self.assertFalse(utils.is_valid_fqdn(input_str))
  402 
  403     def test_valid_host_port_string(self):
  404         valid_pairs = ['10.11.12.13:80',
  405                        '172.17.17.1:65535',
  406                        '[fe80::a:b:c:d]:9990',
  407                        'localhost:9990',
  408                        'localhost.localdomain:9990',
  409                        'glance02.stack42.local:1234',
  410                        'glance04-a.stack47.local:1234',
  411                        'img83.glance.xn--penstack-r74e.org:13080']
  412 
  413         for pair_str in valid_pairs:
  414             host, port = utils.parse_valid_host_port(pair_str)
  415 
  416             escaped = pair_str.startswith('[')
  417             expected_host = '%s%s%s' % ('[' if escaped else '', host,
  418                                         ']' if escaped else '')
  419 
  420             self.assertTrue(pair_str.startswith(expected_host))
  421             self.assertGreater(port, 0)
  422 
  423             expected_pair = '%s:%d' % (expected_host, port)
  424             self.assertEqual(expected_pair, pair_str)
  425 
  426     def test_valid_host_port_string_fail(self):
  427         invalid_pairs = ['',
  428                          '10.11.12.13',
  429                          '172.17.17.1:99999',
  430                          '290.12.52.80:5673',
  431                          'absurd inputs happen',
  432                          u'\u2601',
  433                          u'\u2603:8080',
  434                          'fe80::1',
  435                          '[fe80::2]',
  436                          '<fe80::3>:5673',
  437                          '[fe80::a:b:c:d]9990',
  438                          'fe80:a:b:c:d:e:f:1:2:3:4',
  439                          'fe80:a:b:c:d:e:f:g',
  440                          'fe80::1:8080',
  441                          '[fe80:a:b:c:d:e:f:g]:9090',
  442                          '[a:b:s:u:r:d]:fe80']
  443 
  444         for pair in invalid_pairs:
  445             self.assertRaises(ValueError,
  446                               utils.parse_valid_host_port,
  447                               pair)
  448 
  449     def test_get_stores_from_request_returns_default(self):
  450         enabled_backends = {
  451             "ceph1": "rbd",
  452             "ceph2": "rbd"
  453         }
  454         self.config(enabled_backends=enabled_backends)
  455         store.register_store_opts(CONF)
  456         self.config(default_backend="ceph1", group="glance_store")
  457 
  458         req = webob.Request.blank('/some_request')
  459         mp = "glance.common.utils.glance_store.get_store_from_store_identifier"
  460         with mock.patch(mp) as mock_get_store:
  461             result = utils.get_stores_from_request(req, {})
  462             self.assertEqual(["ceph1"], result)
  463             mock_get_store.assert_called_once_with("ceph1")
  464 
  465     def test_get_stores_from_request_returns_stores_from_body(self):
  466         enabled_backends = {
  467             "ceph1": "rbd",
  468             "ceph2": "rbd"
  469         }
  470         self.config(enabled_backends=enabled_backends)
  471         store.register_store_opts(CONF)
  472         self.config(default_backend="ceph1", group="glance_store")
  473 
  474         body = {"stores": ["ceph1", "ceph2"]}
  475         req = webob.Request.blank("/some_request")
  476         mp = "glance.common.utils.glance_store.get_store_from_store_identifier"
  477         with mock.patch(mp) as mock_get_store:
  478             result = utils.get_stores_from_request(req, body)
  479             self.assertEqual(["ceph1", "ceph2"], result)
  480             mock_get_store.assert_any_call("ceph1")
  481             mock_get_store.assert_any_call("ceph2")
  482             self.assertEqual(mock_get_store.call_count, 2)
  483 
  484     def test_get_stores_from_request_returns_store_from_headers(self):
  485         enabled_backends = {
  486             "ceph1": "rbd",
  487             "ceph2": "rbd"
  488         }
  489         self.config(enabled_backends=enabled_backends)
  490         store.register_store_opts(CONF)
  491         self.config(default_backend="ceph1", group="glance_store")
  492 
  493         headers = {"x-image-meta-store": "ceph2"}
  494         req = webob.Request.blank("/some_request", headers=headers)
  495         mp = "glance.common.utils.glance_store.get_store_from_store_identifier"
  496         with mock.patch(mp) as mock_get_store:
  497             result = utils.get_stores_from_request(req, {})
  498             self.assertEqual(["ceph2"], result)
  499             mock_get_store.assert_called_once_with("ceph2")
  500 
  501     def test_get_stores_from_request_raises_bad_request(self):
  502         enabled_backends = {
  503             "ceph1": "rbd",
  504             "ceph2": "rbd"
  505         }
  506         self.config(enabled_backends=enabled_backends)
  507         store.register_store_opts(CONF)
  508         self.config(default_backend="ceph1", group="glance_store")
  509         headers = {"x-image-meta-store": "ceph2"}
  510         body = {"stores": ["ceph1"]}
  511         req = webob.Request.blank("/some_request", headers=headers)
  512         self.assertRaises(webob.exc.HTTPBadRequest,
  513                           utils.get_stores_from_request, req, body)
  514 
  515     def test_get_stores_from_request_returns_all_stores(self):
  516         enabled_backends = {
  517             "ceph1": "rbd",
  518             "ceph2": "rbd"
  519         }
  520         reserved_stores = {
  521             'os_glance_staging_store': 'file',
  522             'os_glance_tasks_store': 'file'
  523         }
  524         self.config(enabled_backends=enabled_backends)
  525         store.register_store_opts(CONF, reserved_stores=reserved_stores)
  526         self.config(default_backend="ceph1", group="glance_store")
  527         body = {"all_stores": True}
  528         req = webob.Request.blank("/some_request")
  529         mp = "glance.common.utils.glance_store.get_store_from_store_identifier"
  530         with mock.patch(mp) as mock_get_store:
  531             result = sorted(utils.get_stores_from_request(req, body))
  532             self.assertEqual(["ceph1", "ceph2"], result)
  533             mock_get_store.assert_any_call("ceph1")
  534             mock_get_store.assert_any_call("ceph2")
  535             self.assertEqual(mock_get_store.call_count, 2)
  536             self.assertNotIn('os_glance_staging_store', result)
  537             self.assertNotIn('os_glance_tasks_store', result)
  538 
  539     def test_get_stores_from_request_excludes_reserved_stores(self):
  540         enabled_backends = {
  541             "ceph1": "rbd",
  542             "ceph2": "rbd"
  543         }
  544         self.config(enabled_backends=enabled_backends)
  545         store.register_store_opts(CONF)
  546         self.config(default_backend="ceph1", group="glance_store")
  547         body = {"all_stores": True}
  548         req = webob.Request.blank("/some_request")
  549         mp = "glance.common.utils.glance_store.get_store_from_store_identifier"
  550         with mock.patch(mp) as mock_get_store:
  551             result = sorted(utils.get_stores_from_request(req, body))
  552             self.assertEqual(["ceph1", "ceph2"], result)
  553             mock_get_store.assert_any_call("ceph1")
  554             mock_get_store.assert_any_call("ceph2")
  555             self.assertEqual(mock_get_store.call_count, 2)
  556 
  557     def test_get_stores_from_request_excludes_readonly_store(self):
  558         enabled_backends = {
  559             "ceph1": "rbd",
  560             "ceph2": "rbd",
  561             "http": "http"
  562         }
  563         self.config(enabled_backends=enabled_backends)
  564         store.register_store_opts(CONF)
  565         self.config(default_backend="ceph1", group="glance_store")
  566         body = {"all_stores": True}
  567         req = webob.Request.blank("/some_request")
  568         mp = "glance.common.utils.glance_store.get_store_from_store_identifier"
  569         with mock.patch(mp) as mock_get_store:
  570             result = sorted(utils.get_stores_from_request(req, body))
  571             self.assertNotIn("http", result)
  572             self.assertEqual(["ceph1", "ceph2"], result)
  573             mock_get_store.assert_any_call("ceph1")
  574             mock_get_store.assert_any_call("ceph2")
  575             self.assertEqual(mock_get_store.call_count, 2)
  576 
  577     def test_get_stores_from_request_raises_bad_request_with_all_stores(self):
  578         enabled_backends = {
  579             "ceph1": "rbd",
  580             "ceph2": "rbd"
  581         }
  582         self.config(enabled_backends=enabled_backends)
  583         store.register_store_opts(CONF)
  584         self.config(default_backend="ceph1", group="glance_store")
  585         headers = {"x-image-meta-store": "ceph2"}
  586         body = {"stores": ["ceph1"], "all_stores": True}
  587         req = webob.Request.blank("/some_request", headers=headers)
  588         self.assertRaises(webob.exc.HTTPBadRequest,
  589                           utils.get_stores_from_request, req, body)
  590 
  591 
  592 class SplitFilterOpTestCase(test_utils.BaseTestCase):
  593 
  594     def test_less_than_operator(self):
  595         expr = 'lt:bar'
  596         returned = utils.split_filter_op(expr)
  597         self.assertEqual(('lt', 'bar'), returned)
  598 
  599     def test_less_than_equal_operator(self):
  600         expr = 'lte:bar'
  601         returned = utils.split_filter_op(expr)
  602         self.assertEqual(('lte', 'bar'), returned)
  603 
  604     def test_greater_than_operator(self):
  605         expr = 'gt:bar'
  606         returned = utils.split_filter_op(expr)
  607         self.assertEqual(('gt', 'bar'), returned)
  608 
  609     def test_greater_than_equal_operator(self):
  610         expr = 'gte:bar'
  611         returned = utils.split_filter_op(expr)
  612         self.assertEqual(('gte', 'bar'), returned)
  613 
  614     def test_not_equal_operator(self):
  615         expr = 'neq:bar'
  616         returned = utils.split_filter_op(expr)
  617         self.assertEqual(('neq', 'bar'), returned)
  618 
  619     def test_equal_operator(self):
  620         expr = 'eq:bar'
  621         returned = utils.split_filter_op(expr)
  622         self.assertEqual(('eq', 'bar'), returned)
  623 
  624     def test_in_operator(self):
  625         expr = 'in:bar'
  626         returned = utils.split_filter_op(expr)
  627         self.assertEqual(('in', 'bar'), returned)
  628 
  629     def test_split_filter_value_for_quotes(self):
  630         expr = '\"fake\\\"name\",fakename,\"fake,name\"'
  631         returned = utils.split_filter_value_for_quotes(expr)
  632         list_values = ['fake\\"name', 'fakename', 'fake,name']
  633         self.assertEqual(list_values, returned)
  634 
  635     def test_validate_quotes(self):
  636         expr = '\"aaa\\\"aa\",bb,\"cc\"'
  637         returned = utils.validate_quotes(expr)
  638         self.assertIsNone(returned)
  639 
  640         invalid_expr = ['\"aa', 'ss\"', 'aa\"bb\"cc', '\"aa\"\"bb\"']
  641         for expr in invalid_expr:
  642             self.assertRaises(exception.InvalidParameterValue,
  643                               utils.validate_quotes,
  644                               expr)
  645 
  646     def test_default_operator(self):
  647         expr = 'bar'
  648         returned = utils.split_filter_op(expr)
  649         self.assertEqual(('eq', expr), returned)
  650 
  651     def test_default_operator_with_datetime(self):
  652         expr = '2015-08-27T09:49:58Z'
  653         returned = utils.split_filter_op(expr)
  654         self.assertEqual(('eq', expr), returned)
  655 
  656     def test_operator_with_datetime(self):
  657         expr = 'lt:2015-08-27T09:49:58Z'
  658         returned = utils.split_filter_op(expr)
  659         self.assertEqual(('lt', '2015-08-27T09:49:58Z'), returned)
  660 
  661 
  662 class EvaluateFilterOpTestCase(test_utils.BaseTestCase):
  663 
  664     def test_less_than_operator(self):
  665         self.assertTrue(utils.evaluate_filter_op(9, 'lt', 10))
  666         self.assertFalse(utils.evaluate_filter_op(10, 'lt', 10))
  667         self.assertFalse(utils.evaluate_filter_op(11, 'lt', 10))
  668 
  669     def test_less_than_equal_operator(self):
  670         self.assertTrue(utils.evaluate_filter_op(9, 'lte', 10))
  671         self.assertTrue(utils.evaluate_filter_op(10, 'lte', 10))
  672         self.assertFalse(utils.evaluate_filter_op(11, 'lte', 10))
  673 
  674     def test_greater_than_operator(self):
  675         self.assertFalse(utils.evaluate_filter_op(9, 'gt', 10))
  676         self.assertFalse(utils.evaluate_filter_op(10, 'gt', 10))
  677         self.assertTrue(utils.evaluate_filter_op(11, 'gt', 10))
  678 
  679     def test_greater_than_equal_operator(self):
  680         self.assertFalse(utils.evaluate_filter_op(9, 'gte', 10))
  681         self.assertTrue(utils.evaluate_filter_op(10, 'gte', 10))
  682         self.assertTrue(utils.evaluate_filter_op(11, 'gte', 10))
  683 
  684     def test_not_equal_operator(self):
  685         self.assertTrue(utils.evaluate_filter_op(9, 'neq', 10))
  686         self.assertFalse(utils.evaluate_filter_op(10, 'neq', 10))
  687         self.assertTrue(utils.evaluate_filter_op(11, 'neq', 10))
  688 
  689     def test_equal_operator(self):
  690         self.assertFalse(utils.evaluate_filter_op(9, 'eq', 10))
  691         self.assertTrue(utils.evaluate_filter_op(10, 'eq', 10))
  692         self.assertFalse(utils.evaluate_filter_op(11, 'eq', 10))
  693 
  694     def test_invalid_operator(self):
  695         self.assertRaises(exception.InvalidFilterOperatorValue,
  696                           utils.evaluate_filter_op, '10', 'bar', '8')
  697 
  698 
  699 class ImportURITestCase(test_utils.BaseTestCase):
  700 
  701     def test_validate_import_uri(self):
  702         self.assertTrue(utils.validate_import_uri("http://foo.com"))
  703 
  704         self.config(allowed_schemes=['http'],
  705                     group='import_filtering_opts')
  706         self.config(allowed_hosts=['example.com'],
  707                     group='import_filtering_opts')
  708         self.assertTrue(utils.validate_import_uri("http://example.com"))
  709 
  710         self.config(allowed_ports=['8080'],
  711                     group='import_filtering_opts')
  712         self.assertTrue(utils.validate_import_uri("http://example.com:8080"))
  713 
  714     def test_invalid_import_uri(self):
  715         self.assertFalse(utils.validate_import_uri(""))
  716 
  717         self.assertFalse(utils.validate_import_uri("fake_uri"))
  718         self.config(disallowed_schemes=['ftp'],
  719                     group='import_filtering_opts')
  720         self.assertFalse(utils.validate_import_uri("ftp://example.com"))
  721 
  722         self.config(disallowed_hosts=['foo.com'],
  723                     group='import_filtering_opts')
  724         self.assertFalse(utils.validate_import_uri("ftp://foo.com"))
  725 
  726         self.config(disallowed_ports=['8484'],
  727                     group='import_filtering_opts')
  728         self.assertFalse(utils.validate_import_uri("http://localhost:8484"))
  729 
  730     def test_ignored_filtering_options(self):
  731         LOG = logging.getLogger('glance.common.utils')
  732         with mock.patch.object(LOG, 'debug') as mock_run:
  733             self.config(allowed_schemes=['https', 'ftp'],
  734                         group='import_filtering_opts')
  735             self.config(disallowed_schemes=['ftp'],
  736                         group='import_filtering_opts')
  737             self.assertTrue(utils.validate_import_uri("ftp://foo.com"))
  738             mock_run.assert_called_once()
  739         with mock.patch.object(LOG, 'debug') as mock_run:
  740             self.config(allowed_schemes=[],
  741                         group='import_filtering_opts')
  742             self.config(disallowed_schemes=[],
  743                         group='import_filtering_opts')
  744             self.config(allowed_hosts=['example.com', 'foo.com'],
  745                         group='import_filtering_opts')
  746             self.config(disallowed_hosts=['foo.com'],
  747                         group='import_filtering_opts')
  748             self.assertTrue(utils.validate_import_uri("ftp://foo.com"))
  749             mock_run.assert_called_once()
  750         with mock.patch.object(LOG, 'debug') as mock_run:
  751             self.config(allowed_hosts=[],
  752                         group='import_filtering_opts')
  753             self.config(disallowed_hosts=[],
  754                         group='import_filtering_opts')
  755             self.config(allowed_ports=[8080, 8484],
  756                         group='import_filtering_opts')
  757             self.config(disallowed_ports=[8484],
  758                         group='import_filtering_opts')
  759             self.assertTrue(utils.validate_import_uri("ftp://foo.com:8484"))
  760             mock_run.assert_called_once()