# Copyright 2016 Alex Meade # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import copy import datetime import ddt import mock from oslo_config import cfg from oslo_serialization import jsonutils from oslo_utils import uuidutils import six import webob from manila.api.openstack import wsgi from manila.api.v2 import share_group_snapshots from manila.common import constants from manila import context from manila import db from manila import exception from manila import policy from manila import test from manila.tests.api import fakes from manila.tests import db_utils CONF = cfg.CONF @ddt.ddt class ShareGroupSnapshotAPITest(test.TestCase): def setUp(self): super(ShareGroupSnapshotAPITest, self).setUp() self.controller = share_group_snapshots.ShareGroupSnapshotController() self.resource_name = self.controller.resource_name self.api_version = '2.31' self.mock_policy_check = self.mock_object( policy, 'check_policy', mock.Mock(return_value=True)) self.request = fakes.HTTPRequest.blank( '/share-groups', version=self.api_version, experimental=True) self.context = self.request.environ['manila.context'] self.admin_context = context.RequestContext('admin', 'fake', True) self.member_context = context.RequestContext('fake', 'fake') self.flags(transport_url='rabbit://fake:fake@mqhost:5672') def _get_fake_share_group_snapshot(self, **values): snap = { 'id': 'fake_id', 'user_id': 'fakeuser', 'project_id': 'fakeproject', 'status': constants.STATUS_CREATING, 'name': None, 'description': None, 'share_group_id': None, 'created_at': datetime.datetime(1, 1, 1, 1, 1, 1), 'members': [], } snap.update(**values) expected_snap = copy.deepcopy(snap) del expected_snap['user_id'] return snap, expected_snap def _get_fake_simple_share_group_snapshot(self, **values): snap = { 'id': 'fake_id', 'name': None, } snap.update(**values) expected_snap = copy.deepcopy(snap) return snap, expected_snap def _get_fake_share_group_snapshot_member(self, **values): member = { 'id': 'fake_id', 'user_id': 'fakeuser', 'project_id': 'fakeproject', 'status': constants.STATUS_CREATING, 'share_group_snapshot_id': None, 'share_proto': None, 'share_id': None, 'size': None, 'created_at': datetime.datetime(1, 1, 1, 1, 1, 1), } member.update(**values) expected_member = copy.deepcopy(member) del expected_member['user_id'] del expected_member['status'] expected_member['share_protocol'] = member['share_proto'] del expected_member['share_proto'] return member, expected_member def test_create_invalid_body(self): body = {"not_group_snapshot": {}} self.assertRaises( webob.exc.HTTPBadRequest, self.controller.create, self.request, body) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') def test_create_no_share_group_id(self): body = {"share_group_snapshot": {}} self.assertRaises( webob.exc.HTTPBadRequest, self.controller.create, self.request, body) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') def test_create(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() fake_id = six.text_type(uuidutils.generate_uuid()) body = {"share_group_snapshot": {"share_group_id": fake_id}} mock_create = self.mock_object( self.controller.share_group_api, 'create_share_group_snapshot', mock.Mock(return_value=fake_snap)) res_dict = self.controller.create(self.request, body) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') mock_create.assert_called_once_with( self.context, share_group_id=fake_id) res_dict['share_group_snapshot'].pop('links') self.assertEqual(expected_snap, res_dict['share_group_snapshot']) def test_create_group_does_not_exist(self): fake_id = six.text_type(uuidutils.generate_uuid()) body = {"share_group_snapshot": {"share_group_id": fake_id}} self.mock_object( self.controller.share_group_api, 'create_share_group_snapshot', mock.Mock(side_effect=exception.ShareGroupNotFound( share_group_id=six.text_type(uuidutils.generate_uuid())))) self.assertRaises( webob.exc.HTTPBadRequest, self.controller.create, self.request, body) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') def test_create_group_does_not_a_uuid(self): self.mock_object( self.controller.share_group_api, 'create_share_group_snapshot', mock.Mock(side_effect=exception.ShareGroupNotFound( share_group_id='not_a_uuid', ))) body = {"share_group_snapshot": {"share_group_id": "not_a_uuid"}} self.assertRaises( webob.exc.HTTPBadRequest, self.controller.create, self.request, body) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') def test_create_invalid_share_group(self): fake_id = six.text_type(uuidutils.generate_uuid()) body = {"share_group_snapshot": {"share_group_id": fake_id}} self.mock_object( self.controller.share_group_api, 'create_share_group_snapshot', mock.Mock(side_effect=exception.InvalidShareGroup( reason='bad_status'))) self.assertRaises( webob.exc.HTTPConflict, self.controller.create, self.request, body) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') def test_create_with_name(self): fake_name = 'fake_name' fake_snap, expected_snap = self._get_fake_share_group_snapshot( name=fake_name) fake_id = six.text_type(uuidutils.generate_uuid()) mock_create = self.mock_object( self.controller.share_group_api, 'create_share_group_snapshot', mock.Mock(return_value=fake_snap)) body = { "share_group_snapshot": { "share_group_id": fake_id, "name": fake_name, } } res_dict = self.controller.create(self.request, body) res_dict['share_group_snapshot'].pop('links') mock_create.assert_called_once_with( self.context, share_group_id=fake_id, name=fake_name) self.assertEqual(expected_snap, res_dict['share_group_snapshot']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') def test_create_with_description(self): fake_description = 'fake_description' fake_snap, expected_snap = self._get_fake_share_group_snapshot( description=fake_description) fake_id = six.text_type(uuidutils.generate_uuid()) mock_create = self.mock_object( self.controller.share_group_api, 'create_share_group_snapshot', mock.Mock(return_value=fake_snap)) body = { "share_group_snapshot": { "share_group_id": fake_id, "description": fake_description, } } res_dict = self.controller.create(self.request, body) res_dict['share_group_snapshot'].pop('links') mock_create.assert_called_once_with( self.context, share_group_id=fake_id, description=fake_description) self.assertEqual(expected_snap, res_dict['share_group_snapshot']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') def test_create_with_name_and_description(self): fake_name = 'fake_name' fake_description = 'fake_description' fake_id = six.text_type(uuidutils.generate_uuid()) fake_snap, expected_snap = self._get_fake_share_group_snapshot( description=fake_description, name=fake_name) mock_create = self.mock_object( self.controller.share_group_api, 'create_share_group_snapshot', mock.Mock(return_value=fake_snap)) body = { "share_group_snapshot": { "share_group_id": fake_id, "description": fake_description, "name": fake_name, } } res_dict = self.controller.create(self.request, body) res_dict['share_group_snapshot'].pop('links') mock_create.assert_called_once_with( self.context, share_group_id=fake_id, name=fake_name, description=fake_description) self.assertEqual(expected_snap, res_dict['share_group_snapshot']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'create') def test_update_with_name_and_description(self): fake_name = 'fake_name' fake_description = 'fake_description' fake_id = six.text_type(uuidutils.generate_uuid()) fake_snap, expected_snap = self._get_fake_share_group_snapshot( description=fake_description, name=fake_name) self.mock_object( self.controller.share_group_api, 'get_share_group_snapshot', mock.Mock(return_value=fake_snap)) mock_update = self.mock_object( self.controller.share_group_api, 'update_share_group_snapshot', mock.Mock(return_value=fake_snap)) body = { "share_group_snapshot": { "description": fake_description, "name": fake_name, } } res_dict = self.controller.update(self.request, fake_id, body) res_dict['share_group_snapshot'].pop('links') mock_update.assert_called_once_with( self.context, fake_snap, {"name": fake_name, "description": fake_description}) self.assertEqual(expected_snap, res_dict['share_group_snapshot']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'update') def test_update_snapshot_not_found(self): body = {"share_group_snapshot": {}} self.mock_object( self.controller.share_group_api, 'get_share_group_snapshot', mock.Mock(side_effect=exception.NotFound)) self.assertRaises( webob.exc.HTTPNotFound, self.controller.update, self.request, 'fake_id', body) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'update') def test_update_invalid_body(self): body = {"not_group_snapshot": {}} self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update, self.request, 'fake_id', body) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'update') def test_update_invalid_body_invalid_field(self): body = {"share_group_snapshot": {"unknown_field": ""}} exc = self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update, self.request, 'fake_id', body) self.assertIn('unknown_field', six.text_type(exc)) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'update') def test_update_invalid_body_readonly_field(self): body = {"share_group_snapshot": {"created_at": []}} exc = self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update, self.request, 'fake_id', body) self.assertIn('created_at', six.text_type(exc)) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'update') def test_list_index(self): fake_snap, expected_snap = self._get_fake_simple_share_group_snapshot() self.mock_object( self.controller.share_group_api, 'get_all_share_group_snapshots', mock.Mock(return_value=[fake_snap])) res_dict = self.controller.index(self.request) res_dict['share_group_snapshots'][0].pop('links') self.assertEqual([expected_snap], res_dict['share_group_snapshots']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'get_all') def test_list_index_no_share_groups(self): self.mock_object( self.controller.share_group_api, 'get_all_share_group_snapshots', mock.Mock(return_value=[])) res_dict = self.controller.index(self.request) self.assertEqual([], res_dict['share_group_snapshots']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'get_all') def test_list_index_with_limit(self): fake_snap, expected_snap = self._get_fake_simple_share_group_snapshot() fake_snap2, expected_snap2 = ( self._get_fake_simple_share_group_snapshot( id="fake_id2")) self.mock_object( self.controller.share_group_api, 'get_all_share_group_snapshots', mock.Mock(return_value=[fake_snap, fake_snap2])) req = fakes.HTTPRequest.blank('/share-group-snapshots?limit=1', version=self.api_version, experimental=True) req_context = req.environ['manila.context'] res_dict = self.controller.index(req) res_dict['share_group_snapshots'][0].pop('links') self.assertEqual(1, len(res_dict['share_group_snapshots'])) self.assertEqual([expected_snap], res_dict['share_group_snapshots']) self.mock_policy_check.assert_called_once_with( req_context, self.resource_name, 'get_all') def test_list_index_with_limit_and_offset(self): fake_snap, expected_snap = self._get_fake_simple_share_group_snapshot() fake_snap2, expected_snap2 = ( self._get_fake_simple_share_group_snapshot(id="fake_id2")) self.mock_object( self.controller.share_group_api, 'get_all_share_group_snapshots', mock.Mock(return_value=[fake_snap, fake_snap2])) req = fakes.HTTPRequest.blank( '/share-group-snapshots?limit=1&offset=1', version=self.api_version, experimental=True) req_context = req.environ['manila.context'] res_dict = self.controller.index(req) res_dict['share_group_snapshots'][0].pop('links') self.assertEqual(1, len(res_dict['share_group_snapshots'])) self.assertEqual([expected_snap2], res_dict['share_group_snapshots']) self.mock_policy_check.assert_called_once_with( req_context, self.resource_name, 'get_all') def test_list_detail(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() self.mock_object( self.controller.share_group_api, 'get_all_share_group_snapshots', mock.Mock(return_value=[fake_snap])) res_dict = self.controller.detail(self.request) res_dict['share_group_snapshots'][0].pop('links') self.assertEqual(1, len(res_dict['share_group_snapshots'])) self.assertEqual(expected_snap, res_dict['share_group_snapshots'][0]) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'get_all') def test_list_detail_no_share_groups(self): self.mock_object( self.controller.share_group_api, 'get_all_share_group_snapshots', mock.Mock(return_value=[])) res_dict = self.controller.detail(self.request) self.assertEqual([], res_dict['share_group_snapshots']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'get_all') def test_list_detail_with_limit(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() fake_snap2, expected_snap2 = self._get_fake_share_group_snapshot( id="fake_id2") self.mock_object( self.controller.share_group_api, 'get_all_share_group_snapshots', mock.Mock(return_value=[fake_snap, fake_snap2])) req = fakes.HTTPRequest.blank('/share-group-snapshots?limit=1', version=self.api_version, experimental=True) req_context = req.environ['manila.context'] res_dict = self.controller.detail(req) res_dict['share_group_snapshots'][0].pop('links') self.assertEqual(1, len(res_dict['share_group_snapshots'])) self.assertEqual([expected_snap], res_dict['share_group_snapshots']) self.mock_policy_check.assert_called_once_with( req_context, self.resource_name, 'get_all') def test_list_detail_with_limit_and_offset(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() fake_snap2, expected_snap2 = self._get_fake_share_group_snapshot( id="fake_id2") self.mock_object( self.controller.share_group_api, 'get_all_share_group_snapshots', mock.Mock(return_value=[fake_snap, fake_snap2])) req = fakes.HTTPRequest.blank( '/share-group-snapshots?limit=1&offset=1', version=self.api_version, experimental=True) req_context = req.environ['manila.context'] res_dict = self.controller.detail(req) res_dict['share_group_snapshots'][0].pop('links') self.assertEqual(1, len(res_dict['share_group_snapshots'])) self.assertEqual([expected_snap2], res_dict['share_group_snapshots']) self.mock_policy_check.assert_called_once_with( req_context, self.resource_name, 'get_all') def test_delete(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() self.mock_object( self.controller.share_group_api, 'get_share_group_snapshot', mock.Mock(return_value=fake_snap)) self.mock_object( self.controller.share_group_api, 'delete_share_group_snapshot') res = self.controller.delete(self.request, fake_snap['id']) self.assertEqual(202, res.status_code) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'delete') def test_delete_not_found(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() self.mock_object( self.controller.share_group_api, 'get_share_group_snapshot', mock.Mock(side_effect=exception.NotFound)) self.assertRaises( webob.exc.HTTPNotFound, self.controller.delete, self.request, fake_snap['id']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'delete') def test_delete_in_conflicting_status(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() self.mock_object( self.controller.share_group_api, 'get_share_group_snapshot', mock.Mock(return_value=fake_snap)) self.mock_object( self.controller.share_group_api, 'delete_share_group_snapshot', mock.Mock(side_effect=exception.InvalidShareGroupSnapshot( reason='blah'))) self.assertRaises( webob.exc.HTTPConflict, self.controller.delete, self.request, fake_snap['id']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'delete') def test_show(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() self.mock_object( self.controller.share_group_api, 'get_share_group_snapshot', mock.Mock(return_value=fake_snap)) res_dict = self.controller.show(self.request, fake_snap['id']) res_dict['share_group_snapshot'].pop('links') self.assertEqual(expected_snap, res_dict['share_group_snapshot']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'get') def test_show_share_group_not_found(self): fake_snap, expected_snap = self._get_fake_share_group_snapshot() self.mock_object( self.controller.share_group_api, 'get_share_group_snapshot', mock.Mock(side_effect=exception.NotFound)) self.assertRaises( webob.exc.HTTPNotFound, self.controller.show, self.request, fake_snap['id']) self.mock_policy_check.assert_called_once_with( self.context, self.resource_name, 'get') def _get_context(self, role): return getattr(self, '%s_context' % role) def _setup_share_group_snapshot_data(self, share_group_snapshot=None, version='2.31'): if share_group_snapshot is None: share_group_snapshot = db_utils.create_share_group_snapshot( 'fake_id', status=constants.STATUS_AVAILABLE) req = fakes.HTTPRequest.blank( '/v2/fake/share-group-snapshots/%s/action' % share_group_snapshot['id'], version=version) req.headers[wsgi.API_VERSION_REQUEST_HEADER] = version req.headers[wsgi.EXPERIMENTAL_API_REQUEST_HEADER] = 'True' return share_group_snapshot, req @ddt.data(*fakes.fixture_force_delete_with_different_roles) @ddt.unpack def test_share_group_snapshot_force_delete_with_different_roles( self, role, resp_code, version): group_snap, req = self._setup_share_group_snapshot_data() ctxt = self._get_context(role) req.method = 'POST' req.headers['content-type'] = 'application/json' action_name = 'force_delete' body = {action_name: {'status': constants.STATUS_ERROR}} req.body = six.b(jsonutils.dumps(body)) req.headers['X-Openstack-Manila-Api-Version'] = self.api_version req.environ['manila.context'] = ctxt with mock.patch.object( policy, 'check_policy', fakes.mock_fake_admin_check): resp = req.get_response(fakes.app()) # Validate response self.assertEqual(resp_code, resp.status_int) @ddt.data(*fakes.fixture_reset_status_with_different_roles) @ddt.unpack def test_share_group_snapshot_reset_status_with_different_roles( self, role, valid_code, valid_status, version): ctxt = self._get_context(role) group_snap, req = self._setup_share_group_snapshot_data() action_name = 'reset_status' body = {action_name: {'status': constants.STATUS_ERROR}} req.method = 'POST' req.headers['content-type'] = 'application/json' req.body = six.b(jsonutils.dumps(body)) req.headers['X-Openstack-Manila-Api-Version'] = self.api_version req.environ['manila.context'] = ctxt with mock.patch.object( policy, 'check_policy', fakes.mock_fake_admin_check): resp = req.get_response(fakes.app()) # Validate response code and model status self.assertEqual(valid_code, resp.status_int) actual_model = db.share_group_snapshot_get(ctxt, group_snap['id']) self.assertEqual(valid_status, actual_model['status'])