1#   Copyright (c) 2015 Huawei Technologies Co., Ltd.
2#   Copyright (c) 2016 Stratoscale, Ltd.
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
16import mock
17from oslo_config import cfg
18import oslo_messaging as messaging
19from oslo_serialization import jsonutils
20from six.moves import http_client
21from six.moves.urllib.parse import urlencode
22import webob
23
24from cinder.common import constants
25from cinder import context
26from cinder import exception
27from cinder import objects
28from cinder import test
29from cinder.tests.unit.api import fakes
30from cinder.tests.unit import fake_constants as fake
31from cinder.tests.unit import fake_service
32
33CONF = cfg.CONF
34
35
36def app():
37    # no auth, just let environ['cinder.context'] pass through
38    api = fakes.router.APIRouter()
39    mapper = fakes.urlmap.URLMap()
40    mapper['/v2'] = api
41    return mapper
42
43
44def volume_get(self, context, volume_id, viewable_admin_meta=False):
45    if volume_id == fake.VOLUME_ID:
46        return objects.Volume(context, id=fake.VOLUME_ID,
47                              _name_id=fake.VOLUME2_ID,
48                              host='fake_host', cluster_name=None)
49    raise exception.VolumeNotFound(volume_id=volume_id)
50
51
52def api_get_manageable_snapshots(*args, **kwargs):
53    """Replacement for cinder.volume.api.API.get_manageable_snapshots."""
54    snap_id = 'ffffffff-0000-ffff-0000-ffffffffffff'
55    snaps = [
56        {'reference': {'source-name': 'snapshot-%s' % snap_id},
57         'size': 4,
58         'extra_info': 'qos_setting:high',
59         'safe_to_manage': False,
60         'reason_not_safe': 'snapshot in use',
61         'cinder_id': snap_id,
62         'source_reference': {'source-name':
63                              'volume-00000000-ffff-0000-ffff-000000'}},
64        {'reference': {'source-name': 'mysnap'},
65         'size': 5,
66         'extra_info': 'qos_setting:low',
67         'safe_to_manage': True,
68         'reason_not_safe': None,
69         'cinder_id': None,
70         'source_reference': {'source-name': 'myvol'}}]
71    return snaps
72
73
74@mock.patch('cinder.volume.api.API.get', volume_get)
75class SnapshotManageTest(test.TestCase):
76    """Test cases for cinder/api/contrib/snapshot_manage.py
77
78    The API extension adds a POST /os-snapshot-manage API that is passed a
79    cinder volume id, and a driver-specific reference parameter.
80    If everything is passed correctly,
81    then the cinder.volume.api.API.manage_existing_snapshot method
82    is invoked to manage an existing storage object on the host.
83
84    In this set of test cases, we are ensuring that the code correctly parses
85    the request structure and raises the correct exceptions when things are not
86    right, and calls down into cinder.volume.api.API.manage_existing_snapshot
87    with the correct arguments.
88    """
89
90    def setUp(self):
91        super(SnapshotManageTest, self).setUp()
92        self._admin_ctxt = context.RequestContext(fake.USER_ID,
93                                                  fake.PROJECT_ID,
94                                                  is_admin=True)
95        self._non_admin_ctxt = context.RequestContext(fake.USER_ID,
96                                                      fake.PROJECT_ID,
97                                                      is_admin=False)
98
99    def _get_resp_post(self, body):
100        """Helper to execute an os-snapshot-manage API call."""
101        req = webob.Request.blank('/v2/%s/os-snapshot-manage' %
102                                  fake.PROJECT_ID)
103        req.method = 'POST'
104        req.headers['Content-Type'] = 'application/json'
105        req.environ['cinder.context'] = self._admin_ctxt
106        req.body = jsonutils.dump_as_bytes(body)
107        res = req.get_response(app())
108        return res
109
110    @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot')
111    @mock.patch('cinder.volume.api.API.create_snapshot_in_db')
112    @mock.patch('cinder.db.sqlalchemy.api.service_get')
113    def test_manage_snapshot_ok(self, mock_db,
114                                mock_create_snapshot, mock_rpcapi):
115        """Test successful manage snapshot execution.
116
117        Tests for correct operation when valid arguments are passed in the
118        request body. We ensure that cinder.volume.api.API.manage_existing got
119        called with the correct arguments, and that we return the correct HTTP
120        code to the caller.
121        """
122        mock_db.return_value = fake_service.fake_service_obj(
123            self._admin_ctxt,
124            binary=constants.VOLUME_BINARY)
125
126        body = {'snapshot': {'volume_id': fake.VOLUME_ID,
127                             'ref': {'fake_key': 'fake_ref'}}}
128
129        res = self._get_resp_post(body)
130        self.assertEqual(http_client.ACCEPTED, res.status_int, res)
131
132        # Check the db.service_get was called with correct arguments.
133        mock_db.assert_called_once_with(
134            mock.ANY, None, host='fake_host', binary=constants.VOLUME_BINARY,
135            cluster_name=None)
136
137        # Check the create_snapshot_in_db was called with correct arguments.
138        self.assertEqual(1, mock_create_snapshot.call_count)
139        args = mock_create_snapshot.call_args[0]
140        named_args = mock_create_snapshot.call_args[1]
141        self.assertEqual(fake.VOLUME_ID, args[1].get('id'))
142        self.assertTrue(named_args['commit_quota'])
143
144        # Check the volume_rpcapi.manage_existing_snapshot was called with
145        # correct arguments.
146        self.assertEqual(1, mock_rpcapi.call_count)
147        args = mock_rpcapi.call_args[0]
148        self.assertEqual({u'fake_key': u'fake_ref'}, args[2])
149
150    @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot')
151    @mock.patch('cinder.volume.api.API.create_snapshot_in_db')
152    @mock.patch('cinder.objects.service.Service.get_by_id')
153    def test_manage_snapshot_ok_with_metadata_null(
154            self, mock_db, mock_create_snapshot, mock_rpcapi):
155        mock_db.return_value = fake_service.fake_service_obj(
156            self._admin_ctxt,
157            binary=constants.VOLUME_BINARY)
158        body = {'snapshot': {'volume_id': fake.VOLUME_ID,
159                             'ref': {'fake_key': 'fake_ref'},
160                             'name': 'test',
161                             'description': 'test',
162                             'metadata': None}}
163
164        res = self._get_resp_post(body)
165        self.assertEqual(http_client.ACCEPTED, res.status_int)
166        args = mock_create_snapshot.call_args[0]
167        # 5th argument of args is metadata.
168        self.assertIsNone(args[5])
169
170    @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot')
171    @mock.patch('cinder.volume.api.API.create_snapshot_in_db')
172    @mock.patch('cinder.db.sqlalchemy.api.service_get')
173    def test_manage_snapshot_ok_ref_as_string(self, mock_db,
174                                              mock_create_snapshot,
175                                              mock_rpcapi):
176
177        mock_db.return_value = fake_service.fake_service_obj(
178            self._admin_ctxt,
179            binary=constants.VOLUME_BINARY)
180
181        body = {'snapshot': {'volume_id': fake.VOLUME_ID,
182                             'ref': "string"}}
183
184        res = self._get_resp_post(body)
185        self.assertEqual(http_client.ACCEPTED, res.status_int, res)
186
187        # Check the volume_rpcapi.manage_existing_snapshot was called with
188        # correct arguments.
189        self.assertEqual(1, mock_rpcapi.call_count)
190        args = mock_rpcapi.call_args[0]
191        self.assertEqual(body['snapshot']['ref'], args[2])
192
193    @mock.patch('cinder.objects.service.Service.is_up',
194                return_value=True,
195                new_callable=mock.PropertyMock)
196    @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot')
197    @mock.patch('cinder.volume.api.API.create_snapshot_in_db')
198    @mock.patch('cinder.db.sqlalchemy.api.service_get')
199    def test_manage_snapshot_disabled(self, mock_db, mock_create_snapshot,
200                                      mock_rpcapi, mock_is_up):
201        """Test manage snapshot failure due to disabled service."""
202        mock_db.return_value = fake_service.fake_service_obj(self._admin_ctxt,
203                                                             disabled=True)
204        body = {'snapshot': {'volume_id': fake.VOLUME_ID, 'ref': {
205            'fake_key': 'fake_ref'}}}
206        res = self._get_resp_post(body)
207        self.assertEqual(http_client.BAD_REQUEST, res.status_int, res)
208        self.assertEqual(exception.ServiceUnavailable.message,
209                         res.json['badRequest']['message'])
210        mock_create_snapshot.assert_not_called()
211        mock_rpcapi.assert_not_called()
212        mock_is_up.assert_not_called()
213
214    @mock.patch('cinder.objects.service.Service.is_up', return_value=False,
215                new_callable=mock.PropertyMock)
216    @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot')
217    @mock.patch('cinder.volume.api.API.create_snapshot_in_db')
218    @mock.patch('cinder.db.sqlalchemy.api.service_get')
219    def test_manage_snapshot_is_down(self, mock_db, mock_create_snapshot,
220                                     mock_rpcapi, mock_is_up):
221        """Test manage snapshot failure due to down service."""
222        mock_db.return_value = fake_service.fake_service_obj(self._admin_ctxt)
223        body = {'snapshot': {'volume_id': fake.VOLUME_ID,
224                             'ref': {'fake_key': 'fake_ref'}}}
225        res = self._get_resp_post(body)
226        self.assertEqual(http_client.BAD_REQUEST, res.status_int, res)
227        self.assertEqual(exception.ServiceUnavailable.message,
228                         res.json['badRequest']['message'])
229        mock_create_snapshot.assert_not_called()
230        mock_rpcapi.assert_not_called()
231        self.assertTrue(mock_is_up.called)
232
233    def test_manage_snapshot_missing_volume_id(self):
234        """Test correct failure when volume_id is not specified."""
235        body = {'snapshot': {'ref': 'fake_ref'}}
236        res = self._get_resp_post(body)
237        self.assertEqual(http_client.BAD_REQUEST, res.status_int)
238
239    def test_manage_snapshot_missing_ref(self):
240        """Test correct failure when the ref is not specified."""
241        body = {'snapshot': {'volume_id': fake.VOLUME_ID}}
242        res = self._get_resp_post(body)
243        self.assertEqual(http_client.BAD_REQUEST, res.status_int)
244
245    def test_manage_snapshot_error_body(self):
246        """Test correct failure when body is invaild."""
247        body = {'error_snapshot': {'volume_id': fake.VOLUME_ID}}
248        res = self._get_resp_post(body)
249        self.assertEqual(http_client.BAD_REQUEST, res.status_int)
250
251    def test_manage_snapshot_error_volume_id(self):
252        """Test correct failure when volume can't be found."""
253        body = {'snapshot': {'volume_id': 'error_volume_id', 'ref': {}}}
254        res = self._get_resp_post(body)
255        self.assertEqual(http_client.BAD_REQUEST, res.status_int)
256
257    def _get_resp_get(self, host, detailed, paging, admin=True):
258        """Helper to execute a GET os-snapshot-manage API call."""
259        params = {'host': host}
260        if paging:
261            params.update({'marker': '1234', 'limit': 10,
262                           'offset': 4, 'sort': 'reference:asc'})
263        query_string = "?%s" % urlencode(params)
264        detail = ""
265        if detailed:
266            detail = "/detail"
267        url = "/v2/%s/os-snapshot-manage%s%s" % (fake.PROJECT_ID, detail,
268                                                 query_string)
269        req = webob.Request.blank(url)
270        req.method = 'GET'
271        req.headers['Content-Type'] = 'application/json'
272        req.environ['cinder.context'] = (self._admin_ctxt if admin
273                                         else self._non_admin_ctxt)
274        res = req.get_response(app())
275        return res
276
277    @mock.patch('cinder.volume.api.API.get_manageable_snapshots',
278                wraps=api_get_manageable_snapshots)
279    def test_get_manageable_snapshots_non_admin(self, mock_api_manageable):
280        res = self._get_resp_get('fakehost', False, False, admin=False)
281        self.assertEqual(http_client.FORBIDDEN, res.status_int)
282        self.assertEqual(False, mock_api_manageable.called)
283        res = self._get_resp_get('fakehost', True, False, admin=False)
284        self.assertEqual(http_client.FORBIDDEN, res.status_int)
285        self.assertEqual(False, mock_api_manageable.called)
286
287    @mock.patch('cinder.volume.api.API.get_manageable_snapshots',
288                wraps=api_get_manageable_snapshots)
289    def test_get_manageable_snapshots_ok(self, mock_api_manageable):
290        res = self._get_resp_get('fakehost', False, False)
291        snap_name = 'snapshot-ffffffff-0000-ffff-0000-ffffffffffff'
292        exp = {'manageable-snapshots':
293               [{'reference': {'source-name': snap_name}, 'size': 4,
294                 'safe_to_manage': False,
295                 'source_reference':
296                 {'source-name': 'volume-00000000-ffff-0000-ffff-000000'}},
297                {'reference': {'source-name': 'mysnap'}, 'size': 5,
298                 'safe_to_manage': True,
299                 'source_reference': {'source-name': 'myvol'}}]}
300        self.assertEqual(http_client.OK, res.status_int)
301        self.assertEqual(jsonutils.loads(res.body), exp)
302        mock_api_manageable.assert_called_once_with(
303            self._admin_ctxt, 'fakehost', None, limit=CONF.osapi_max_limit,
304            marker=None, offset=0, sort_dirs=['desc'],
305            sort_keys=['reference'])
306
307    @mock.patch('cinder.volume.api.API.get_manageable_snapshots',
308                side_effect=messaging.RemoteError(
309                    exc_type='InvalidInput', value='marker not found: 1234'))
310    def test_get_manageable_snapshots_non_existent_marker(
311            self, mock_api_manageable):
312        res = self._get_resp_get('fakehost', detailed=False, paging=True)
313        self.assertEqual(http_client.BAD_REQUEST, res.status_int)
314        self.assertTrue(mock_api_manageable.called)
315
316    @mock.patch('cinder.volume.api.API.get_manageable_snapshots',
317                wraps=api_get_manageable_snapshots)
318    def test_get_manageable_snapshots_detailed_ok(self, mock_api_manageable):
319        res = self._get_resp_get('fakehost', True, True)
320        snap_id = 'ffffffff-0000-ffff-0000-ffffffffffff'
321        exp = {'manageable-snapshots':
322               [{'reference': {'source-name': 'snapshot-%s' % snap_id},
323                 'size': 4, 'safe_to_manage': False, 'cinder_id': snap_id,
324                 'reason_not_safe': 'snapshot in use',
325                 'extra_info': 'qos_setting:high',
326                 'source_reference':
327                 {'source-name': 'volume-00000000-ffff-0000-ffff-000000'}},
328                {'reference': {'source-name': 'mysnap'}, 'size': 5,
329                 'cinder_id': None, 'safe_to_manage': True,
330                 'reason_not_safe': None, 'extra_info': 'qos_setting:low',
331                 'source_reference': {'source-name': 'myvol'}}]}
332        self.assertEqual(http_client.OK, res.status_int)
333        self.assertEqual(jsonutils.loads(res.body), exp)
334        mock_api_manageable.assert_called_once_with(
335            self._admin_ctxt, 'fakehost', None, limit=10, marker='1234',
336            offset=4, sort_dirs=['asc'], sort_keys=['reference'])
337
338    @mock.patch('cinder.volume.api.API.get_manageable_snapshots',
339                side_effect=messaging.RemoteError(
340                    exc_type='InvalidInput', value='marker not found: 1234'))
341    def test_get_manageable_snapshots_non_existent_marker_detailed(
342            self, mock_api_manageable):
343        res = self._get_resp_get('fakehost', detailed=True, paging=True)
344        self.assertEqual(http_client.BAD_REQUEST, res.status_int)
345        self.assertTrue(mock_api_manageable.called)
346
347    @mock.patch('cinder.objects.service.Service.is_up', return_value=True)
348    @mock.patch('cinder.db.sqlalchemy.api.service_get')
349    def test_get_manageable_snapshots_disabled(self, mock_db, mock_is_up):
350        mock_db.return_value = fake_service.fake_service_obj(self._admin_ctxt,
351                                                             disabled=True)
352        res = self._get_resp_get('host_ok', False, True)
353        self.assertEqual(http_client.BAD_REQUEST, res.status_int, res)
354        self.assertEqual(exception.ServiceUnavailable.message,
355                         res.json['badRequest']['message'])
356        mock_is_up.assert_not_called()
357
358    @mock.patch('cinder.objects.service.Service.is_up', return_value=False,
359                new_callable=mock.PropertyMock)
360    @mock.patch('cinder.db.sqlalchemy.api.service_get')
361    def test_get_manageable_snapshots_is_down(self, mock_db, mock_is_up):
362        mock_db.return_value = fake_service.fake_service_obj(self._admin_ctxt)
363        res = self._get_resp_get('host_ok', False, True)
364        self.assertEqual(http_client.BAD_REQUEST, res.status_int, res)
365        self.assertEqual(exception.ServiceUnavailable.message,
366                         res.json['badRequest']['message'])
367        self.assertTrue(mock_is_up.called)
368
369    @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot')
370    @mock.patch('cinder.volume.api.API.create_snapshot_in_db')
371    @mock.patch('cinder.objects.service.Service.get_by_id')
372    def test_manage_snapshot_with_null_validate(
373            self, mock_db, mock_create_snapshot, mock_rpcapi):
374        mock_db.return_value = fake_service.fake_service_obj(
375            self._admin_ctxt,
376            binary=constants.VOLUME_BINARY)
377        body = {'snapshot': {'volume_id': fake.VOLUME_ID,
378                             'ref': {'fake_key': 'fake_ref'},
379                             'name': None,
380                             'description': None}}
381
382        res = self._get_resp_post(body)
383        self.assertEqual(http_client.ACCEPTED, res.status_int, res)
384        self.assertIn('snapshot', jsonutils.loads(res.body))
385