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