1# Copyright (c) 2016 Stratoscale, Ltd. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); you may 4# not use this file except in compliance with the License. You may obtain 5# a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 12# License for the specific language governing permissions and limitations 13# under the License. 14 15import ddt 16import mock 17from oslo_config import cfg 18from oslo_serialization import jsonutils 19from six.moves import http_client 20from six.moves.urllib.parse import urlencode 21import webob 22 23from cinder.api import microversions as mv 24from cinder.api.v3 import router as router_v3 25from cinder.common import constants 26from cinder import context 27from cinder import objects 28from cinder import test 29from cinder.tests.unit.api.contrib import test_volume_manage as test_contrib 30from cinder.tests.unit.api import fakes 31from cinder.tests.unit import fake_constants as fake 32 33 34CONF = cfg.CONF 35 36 37def app(): 38 # no auth, just let environ['cinder.context'] pass through 39 api = router_v3.APIRouter() 40 mapper = fakes.urlmap.URLMap() 41 mapper['/v3'] = api 42 return mapper 43 44 45@ddt.ddt 46@mock.patch('cinder.objects.service.Service.get_by_host_and_topic', 47 test_contrib.service_get) 48@mock.patch('cinder.volume.volume_types.get_volume_type_by_name', 49 test_contrib.vt_get_volume_type_by_name) 50@mock.patch('cinder.volume.volume_types.get_volume_type', 51 test_contrib.vt_get_volume_type) 52class VolumeManageTest(test.TestCase): 53 """Test cases for cinder/api/v3/volume_manage.py""" 54 55 def setUp(self): 56 super(VolumeManageTest, self).setUp() 57 self._admin_ctxt = context.RequestContext(fake.USER_ID, 58 fake.PROJECT_ID, 59 True) 60 61 def _get_resp_post(self, body, version=mv.MANAGE_EXISTING_LIST): 62 """Helper to execute a POST manageable_volumes API call.""" 63 req = webob.Request.blank('/v3/%s/manageable_volumes' % 64 fake.PROJECT_ID) 65 req.method = 'POST' 66 req.headers = mv.get_mv_header(version) 67 req.headers['Content-Type'] = 'application/json' 68 req.environ['cinder.context'] = self._admin_ctxt 69 req.body = jsonutils.dump_as_bytes(body) 70 res = req.get_response(app()) 71 return res 72 73 @mock.patch('cinder.volume.api.API.manage_existing', 74 wraps=test_contrib.api_manage) 75 @mock.patch( 76 'cinder.api.openstack.wsgi.Controller.validate_name_and_description') 77 def test_manage_volume_route(self, mock_validate, mock_api_manage): 78 """Test call to manage volume. 79 80 There is currently no change between the API in contrib and the API in 81 v3, so here we simply check that the call is routed properly, rather 82 than copying all the tests. 83 """ 84 body = {'volume': {'host': 'host_ok', 'ref': 'fake_ref'}} 85 res = self._get_resp_post(body) 86 self.assertEqual(http_client.ACCEPTED, res.status_int, res) 87 88 def test_manage_volume_previous_version(self): 89 body = {'volume': {'host': 'host_ok', 'ref': 'fake_ref'}} 90 res = self._get_resp_post(body) 91 self.assertEqual(http_client.BAD_REQUEST, res.status_int, res) 92 93 def _get_resp_get(self, host, detailed, paging, 94 version=mv.MANAGE_EXISTING_LIST, **kwargs): 95 """Helper to execute a GET os-volume-manage API call.""" 96 params = {'host': host} if host else {} 97 params.update(kwargs) 98 if paging: 99 params.update({'marker': '1234', 'limit': 10, 100 'offset': 4, 'sort': 'reference:asc'}) 101 query_string = "?%s" % urlencode(params) 102 detail = "" 103 if detailed: 104 detail = "/detail" 105 106 req = webob.Request.blank('/v3/%s/manageable_volumes%s%s' % 107 (fake.PROJECT_ID, detail, query_string)) 108 req.method = 'GET' 109 req.headers = mv.get_mv_header(version) 110 req.headers['Content-Type'] = 'application/json' 111 req.environ['cinder.context'] = self._admin_ctxt 112 res = req.get_response(app()) 113 return res 114 115 @mock.patch('cinder.volume.api.API.get_manageable_volumes', 116 wraps=test_contrib.api_get_manageable_volumes) 117 def test_get_manageable_volumes_route(self, mock_api_manageable): 118 """Test call to get manageable volumes. 119 120 There is currently no change between the API in contrib and the API in 121 v3, so here we simply check that the call is routed properly, rather 122 than copying all the tests. 123 """ 124 res = self._get_resp_get('fakehost', False, True) 125 self.assertEqual(http_client.OK, res.status_int) 126 127 def test_get_manageable_volumes_previous_version(self): 128 res = self._get_resp_get( 129 'fakehost', False, True, 130 version=mv.get_prior_version(mv.MANAGE_EXISTING_LIST)) 131 self.assertEqual(http_client.NOT_FOUND, res.status_int) 132 133 @mock.patch('cinder.volume.api.API.get_manageable_volumes', 134 wraps=test_contrib.api_get_manageable_volumes) 135 def test_get_manageable_volumes_detail_route(self, mock_api_manageable): 136 """Test call to get manageable volumes (detailed). 137 138 There is currently no change between the API in contrib and the API in 139 v3, so here we simply check that the call is routed properly, rather 140 than copying all the tests. 141 """ 142 res = self._get_resp_get('fakehost', True, False) 143 self.assertEqual(http_client.OK, res.status_int) 144 145 def test_get_manageable_volumes_detail_previous_version(self): 146 res = self._get_resp_get( 147 'fakehost', True, False, 148 version=mv.get_prior_version(mv.MANAGE_EXISTING_LIST)) 149 self.assertEqual(http_client.NOT_FOUND, res.status_int) 150 151 @ddt.data((True, True, 'detail_list'), (True, False, 'summary_list'), 152 (False, True, 'detail_list'), (False, False, 'summary_list')) 153 @ddt.unpack 154 @mock.patch('cinder.objects.Service.is_up', True) 155 @mock.patch('cinder.volume.rpcapi.VolumeAPI._get_cctxt') 156 @mock.patch('cinder.objects.Service.get_by_id') 157 def test_get_manageable_detail(self, clustered, is_detail, view_method, 158 get_service_mock, get_cctxt_mock): 159 if clustered: 160 host = None 161 cluster_name = 'mycluster' 162 version = mv.MANAGE_EXISTING_CLUSTER 163 kwargs = {'cluster': cluster_name} 164 else: 165 host = 'fakehost' 166 cluster_name = None 167 version = mv.MANAGE_EXISTING_LIST 168 kwargs = {} 169 service = objects.Service(disabled=False, host='fakehost', 170 cluster_name=cluster_name) 171 get_service_mock.return_value = service 172 volumes = [mock.sentinel.volume1, mock.sentinel.volume2] 173 get_cctxt_mock.return_value.call.return_value = volumes 174 175 view_data = {'manageable-volumes': [{'vol': str(v)} for v in volumes]} 176 view_path = ('cinder.api.views.manageable_volumes.ViewBuilder.' + 177 view_method) 178 with mock.patch(view_path, return_value=view_data) as detail_view_mock: 179 res = self._get_resp_get(host, is_detail, False, version=version, 180 **kwargs) 181 182 self.assertEqual(http_client.OK, res.status_int) 183 get_cctxt_mock.assert_called_once_with(service.service_topic_queue, 184 version=('3.10', '3.0')) 185 get_cctxt_mock.return_value.call.assert_called_once_with( 186 mock.ANY, 'get_manageable_volumes', marker=None, 187 limit=CONF.osapi_max_limit, offset=0, sort_keys=['reference'], 188 sort_dirs=['desc'], want_objects=True) 189 detail_view_mock.assert_called_once_with(mock.ANY, volumes, 190 len(volumes)) 191 get_service_mock.assert_called_once_with( 192 mock.ANY, None, host=host, binary=constants.VOLUME_BINARY, 193 cluster_name=cluster_name) 194 195 @ddt.data(mv.MANAGE_EXISTING_LIST, mv.MANAGE_EXISTING_CLUSTER) 196 def test_get_manageable_missing_host(self, version): 197 res = self._get_resp_get(None, True, False, version=version) 198 self.assertEqual(http_client.BAD_REQUEST, res.status_int) 199 200 def test_get_manageable_both_host_cluster(self): 201 res = self._get_resp_get('host', True, False, 202 version=mv.MANAGE_EXISTING_CLUSTER, 203 cluster='cluster') 204 self.assertEqual(http_client.BAD_REQUEST, res.status_int) 205