"Fossies" - the Fresh Open Source Software Archive

Member "zun-4.0.0/zun/tests/unit/api/controllers/v1/test_images.py" (16 Oct 2019, 11112 Bytes) of package /linux/misc/openstack/zun-4.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 last Fossies "Diffs" side-by-side code changes report for "test_images.py": 2.1.0_vs_3.0.0.

    1 # Licensed under the Apache License, Version 2.0 (the "License");
    2 #    you may not use this file except in compliance with the License.
    3 #    You may obtain a copy of the License at
    4 #
    5 #        http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 #    Unless required by applicable law or agreed to in writing, software
    8 #    distributed under the License is distributed on an "AS IS" BASIS,
    9 #    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   10 #    See the License for the specific language governing permissions and
   11 #    limitations under the License.
   12 
   13 import mock
   14 from mock import patch
   15 from webtest.app import AppError
   16 
   17 from oslo_utils import uuidutils
   18 
   19 from zun.common import exception
   20 import zun.conf
   21 from zun import objects
   22 from zun.tests.unit.api import base as api_base
   23 from zun.tests.unit.db import utils
   24 
   25 
   26 CONF = zun.conf.CONF
   27 
   28 
   29 class TestImageController(api_base.FunctionalTest):
   30     @mock.patch('zun.common.policy.enforce', return_value=True)
   31     @patch('zun.compute.api.API.image_pull')
   32     @patch('zun.objects.ComputeNode.get_by_name')
   33     def test_image_pull(self, mock_get, mock_image_pull, mock_policy_enforce):
   34         mock_image_pull.side_effect = lambda x, y: y
   35         mock_get.return_value = mock.Mock(hostname='fake-host')
   36 
   37         params = ('{"repo": "hello-world", "host": "fake-host"}')
   38         response = self.post('/v1/images/',
   39                              params=params,
   40                              content_type='application/json')
   41 
   42         self.assertEqual(202, response.status_int)
   43         self.assertTrue(mock_image_pull.called)
   44 
   45         params = ('{"repo": "hello-world:test", "host": "fake-host"}')
   46         response = self.post('/v1/images/',
   47                              params=params,
   48                              content_type='application/json')
   49 
   50         self.assertEqual(202, response.status_int)
   51         self.assertTrue(mock_image_pull.called)
   52 
   53     @patch('zun.compute.api.API.image_pull')
   54     def test_image_pull_with_no_repo(self, mock_image_pull):
   55         params = {}
   56         with self.assertRaisesRegex(AppError,
   57                                     "is a required property"):
   58             self.post('/v1/images/',
   59                       params=params,
   60                       content_type='application/json')
   61         self.assertTrue(mock_image_pull.not_called)
   62 
   63     @mock.patch('zun.common.policy.enforce', return_value=True)
   64     @patch('zun.compute.api.API.image_pull')
   65     @patch('zun.objects.ComputeNode.get_by_name')
   66     def test_image_pull_conflict(
   67             self, mock_get, mock_image_pull, mock_policy_enforce):
   68         mock_image_pull.side_effect = lambda x, y: y
   69         mock_get.return_value = mock.Mock(hostname='fake-host')
   70 
   71         params = ('{"repo": "hello-world", "host": "fake-host"}')
   72         response = self.post('/v1/images/',
   73                              params=params,
   74                              content_type='application/json')
   75 
   76         self.assertEqual(202, response.status_int)
   77         self.assertTrue(mock_image_pull.called)
   78         self.assertRaises(AppError, self.post, '/v1/images/',
   79                           params=params, content_type='application/json')
   80         self.assertTrue(mock_image_pull.not_called)
   81 
   82     @mock.patch('zun.common.policy.enforce', return_value=True)
   83     @patch('zun.compute.api.API.image_pull')
   84     @patch('zun.objects.ComputeNode.get_by_name')
   85     def test_pull_image_set_project_id_and_user_id(
   86             self, mock_get, mock_image_pull, mock_policy_enforce):
   87         def _create_side_effect(cnxt, image):
   88             self.assertEqual(self.context.project_id, image.project_id)
   89             self.assertEqual(self.context.user_id, image.user_id)
   90             return image
   91         mock_image_pull.side_effect = _create_side_effect
   92         mock_get.return_value = mock.Mock(hostname='fake-host')
   93 
   94         params = ('{"repo": "hello-world", "host": "foo"}')
   95         self.post('/v1/images/',
   96                   params=params,
   97                   content_type='application/json')
   98 
   99     @mock.patch('zun.common.policy.enforce', return_value=True)
  100     @patch('zun.compute.api.API.image_pull')
  101     @patch('zun.objects.ComputeNode.get_by_name')
  102     def test_image_pull_with_tag(
  103             self, mock_get, mock_image_pull, mock_policy_enforce):
  104         mock_image_pull.side_effect = lambda x, y: y
  105         mock_get.return_value = mock.Mock(hostname='fake-host')
  106 
  107         params = ('{"repo": "hello-world:latest", "host": "fake-host"}')
  108         response = self.post('/v1/images/',
  109                              params=params,
  110                              content_type='application/json')
  111 
  112         self.assertEqual(202, response.status_int)
  113         self.assertTrue(mock_image_pull.called)
  114 
  115     @mock.patch('zun.common.policy.enforce', return_value=True)
  116     @patch('zun.objects.Image.list')
  117     def test_get_all_images(self, mock_image_list, mock_policy_enforce):
  118         test_image = utils.get_test_image()
  119         images = [objects.Image(self.context, **test_image)]
  120         mock_image_list.return_value = images
  121 
  122         response = self.get('/v1/images/')
  123 
  124         mock_image_list.assert_called_once_with(mock.ANY,
  125                                                 1000, None, 'id', 'asc',
  126                                                 filters=None)
  127         self.assertEqual(200, response.status_int)
  128         actual_images = response.json['images']
  129         self.assertEqual(1, len(actual_images))
  130         self.assertEqual(test_image['uuid'],
  131                          actual_images[0].get('uuid'))
  132         self.assertEqual(test_image['host'],
  133                          actual_images[0].get('host'))
  134 
  135     @patch('zun.common.policy.enforce')
  136     @patch('zun.objects.Image.get_by_uuid')
  137     def test_get_one_image_by_uuid(self, mock_image_get_by_uuid, mock_policy):
  138         mock_policy.return_value = True
  139         test_image = utils.get_test_image()
  140         test_image_obj = objects.Image(self.context, **test_image)
  141         mock_image_get_by_uuid.return_value = test_image_obj
  142 
  143         response = self.get('/v1/images/%s/' % test_image['uuid'])
  144         mock_image_get_by_uuid.assert_called_once_with(
  145             mock.ANY,
  146             test_image['uuid'])
  147         self.assertEqual(200, response.status_int)
  148         self.assertEqual(test_image['uuid'],
  149                          response.json['uuid'])
  150         self.assertEqual(test_image['host'],
  151                          response.json['host'])
  152 
  153     @mock.patch('zun.common.policy.enforce', return_value=True)
  154     @patch('zun.objects.Image.list')
  155     def test_get_all_images_with_pagination_marker(
  156             self, mock_image_list, mock_policy_enforce):
  157         image_list = []
  158         for id_ in range(4):
  159             test_image = utils.create_test_image(
  160                 context=self.context,
  161                 id=id_,
  162                 repo='testrepo' + str(id_),
  163                 uuid=uuidutils.generate_uuid())
  164             image_list.append(objects.Image(self.context, **test_image))
  165         mock_image_list.return_value = image_list[-1:]
  166         response = self.get('/v1/images/?limit=3&marker=%s'
  167                             % image_list[2].uuid)
  168 
  169         self.assertEqual(200, response.status_int)
  170         actual_images = response.json['images']
  171         self.assertEqual(1, len(actual_images))
  172         self.assertEqual(image_list[-1].uuid,
  173                          actual_images[0].get('uuid'))
  174 
  175     @patch('zun.compute.api.API.image_search')
  176     def test_search_image(self, mock_image_search):
  177         mock_image_search.return_value = {'name': 'redis', 'stars': 2000}
  178         response = self.get('/v1/images/redis/search/')
  179         self.assertEqual(200, response.status_int)
  180         mock_image_search.assert_called_once_with(
  181             mock.ANY, 'redis', CONF.default_image_driver, False)
  182 
  183     @patch('zun.compute.api.API.image_search')
  184     def test_search_image_with_tag(self, mock_image_search):
  185         mock_image_search.return_value = {'name': 'redis', 'stars': 2000}
  186         response = self.get('/v1/images/redis:test/search/')
  187         self.assertEqual(200, response.status_int)
  188         mock_image_search.assert_called_once_with(
  189             mock.ANY, 'redis:test', CONF.default_image_driver, False)
  190 
  191     @patch('zun.compute.api.API.image_search')
  192     def test_search_image_not_found(self, mock_image_search):
  193         mock_image_search.side_effect = exception.ImageNotFound
  194         self.assertRaises(AppError, self.get, '/v1/images/redis/search/')
  195         mock_image_search.assert_called_once_with(
  196             mock.ANY, 'redis', CONF.default_image_driver, False)
  197 
  198     @patch('zun.compute.rpcapi.API.image_search')
  199     def test_search_image_with_exact_match_true(self, mock_image_search):
  200         mock_image_search.return_value = {'name': 'redis', 'stars': 2000}
  201         response = self.get(
  202             '/v1/images/redis/search?exact_match=true&image_driver=docker')
  203         self.assertEqual(200, response.status_int)
  204         mock_image_search.assert_called_once_with(
  205             mock.ANY, 'redis', 'docker', True)
  206 
  207     @patch('zun.compute.rpcapi.API.image_search')
  208     def test_search_image_with_exact_match_false(self, mock_image_search):
  209         mock_image_search.return_value = {'name': 'redis', 'stars': 2000}
  210         response = self.get(
  211             '/v1/images/redis/search?exact_match=false&image_driver=glance')
  212         self.assertEqual(200, response.status_int)
  213         mock_image_search.assert_called_once_with(
  214             mock.ANY, 'redis', 'glance', False)
  215 
  216     @patch('zun.compute.api.API.image_search')
  217     def test_search_image_with_exact_match_wrong(self, mock_image_search):
  218         mock_image_search.side_effect = exception.InvalidValue
  219         with self.assertRaisesRegex(AppError,
  220                                     "Invalid input for query parameters"):
  221             self.get('/v1/images/redis/search?exact_match=wrong')
  222 
  223     @patch('zun.compute.api.API.image_search')
  224     def test_search_image_with_image_driver_wrong(self, mock_image_search):
  225         mock_image_search.side_effect = exception.InvalidValue
  226         with self.assertRaisesRegex(AppError,
  227                                     "Invalid input for query parameters"):
  228             self.get('/v1/images/redis/search?image_driver=wrong')
  229 
  230 
  231 class TestImageEnforcement(api_base.FunctionalTest):
  232 
  233     def _common_policy_check(self, rule, func, *arg, **kwarg):
  234         self.policy.set_rules({rule: 'project_id:non_fake'})
  235         response = func(*arg, **kwarg)
  236         self.assertEqual(403, response.status_int)
  237         self.assertEqual('application/json', response.content_type)
  238         self.assertTrue(
  239             "Policy doesn't allow %s to be performed." % rule,
  240             response.json['errors'][0]['detail'])
  241 
  242     def test_policy_disallow_get_all(self):
  243         self._common_policy_check(
  244             'image:get_all', self.get_json, '/images/',
  245             expect_errors=True)
  246 
  247     def test_policy_disallow_create(self):
  248         params = ('{"repo": "foo", "host": "foo"}')
  249         self._common_policy_check(
  250             'image:pull', self.post, '/v1/images/',
  251             params=params,
  252             content_type='application/json',
  253             expect_errors=True)