Coverage for manila/tests/api/v2/test_share_snapshot_instances.py: 99%

1# Copyright 2016 Huawei Inc. 

2# All Rights Reserved. 

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 

15from unittest import mock 

16 

17import ddt 

18from oslo_config import cfg 

19from oslo_serialization import jsonutils 

20from webob import exc 

21 

22from manila.api.v2 import share_snapshot_instances 

23from manila.common import constants 

24from manila import context 

25from manila import exception 

26from manila import policy 

27from manila import test 

28from manila.tests.api import fakes 

29from manila.tests import db_utils 

30from manila.tests import fake_share 

31 

32CONF = cfg.CONF 

33 

34 

35@ddt.ddt 

36class ShareSnapshotInstancesApiTest(test.TestCase): 

37 """Share snapshot instance Api Test.""" 

38 def setUp(self): 

39 super(ShareSnapshotInstancesApiTest, self).setUp() 

40 self.controller = (share_snapshot_instances. 

41 ShareSnapshotInstancesController()) 

42 self.resource_name = self.controller.resource_name 

43 self.api_version = '2.19' 

44 self.snapshot_instances_req = fakes.HTTPRequest.blank( 

45 '/snapshot-instances', version=self.api_version) 

46 self.admin_context = context.RequestContext('admin', 'fake', True) 

47 self.member_context = context.RequestContext('fake', 'fake') 

48 self.snapshot_instances_req.environ['manila.context'] = ( 

49 self.admin_context) 

50 self.snapshot_instances_req_admin = fakes.HTTPRequest.blank( 

51 '/snapshot-instances', version=self.api_version, 

52 use_admin_context=True) 

53 self.mock_policy_check = self.mock_object(policy, 'check_policy') 

54 

55 def _get_fake_snapshot_instance(self, summary=False, **values): 

56 snapshot_instance = fake_share.fake_snapshot_instance( 

57 as_primitive=True) 

58 expected_keys = { 

59 'id', 

60 'snapshot_id', 

61 'status', 

62 } 

63 expected_snapshot_instance = {key: snapshot_instance[key] for key 

64 in snapshot_instance if key 

65 in expected_keys} 

66 

67 if not summary: 

68 expected_snapshot_instance['share_id'] = ( 

69 snapshot_instance.get('share_instance').get('share_id')) 

70 expected_snapshot_instance.update({ 

71 'created_at': snapshot_instance.get('created_at'), 

72 'updated_at': snapshot_instance.get('updated_at'), 

73 'progress': snapshot_instance.get('progress'), 

74 'provider_location': snapshot_instance.get( 

75 'provider_location'), 

76 'share_instance_id': snapshot_instance.get( 

77 'share_instance_id'), 

78 }) 

79 

80 return snapshot_instance, expected_snapshot_instance 

81 

82 def _setup_snapshot_instance_data(self, instance=None): 

83 if instance is None: 83 ↛ 92line 83 didn't jump to line 92 because the condition on line 83 was always true

84 share_instance = db_utils.create_share_instance( 

85 status=constants.STATUS_AVAILABLE, 

86 share_id='fake_share_id_1') 

87 instance = db_utils.create_snapshot_instance( 

88 'fake_snapshot_id_1', 

89 status=constants.STATUS_AVAILABLE, 

90 share_instance_id=share_instance['id']) 

91 

92 path = '/v2/fake/snapshot-instances/%s/action' % instance['id'] 

93 req = fakes.HTTPRequest.blank(path, version=self.api_version, 

94 script_name=path) 

95 req.method = 'POST' 

96 req.headers['content-type'] = 'application/json' 

97 req.headers['X-Openstack-Manila-Api-Version'] = self.api_version 

98 

99 return instance, req 

100 

101 def _get_context(self, role): 

102 return getattr(self, '%s_context' % role) 

103 

104 @ddt.data(None, 'FAKE_SNAPSHOT_ID') 

105 def test_list_snapshot_instances_summary(self, snapshot_id): 

106 snapshot_instance, expected_snapshot_instance = ( 

107 self._get_fake_snapshot_instance(summary=True)) 

108 self.mock_object(share_snapshot_instances.db, 

109 'share_snapshot_instance_get_all_with_filters', 

110 mock.Mock(return_value=[snapshot_instance])) 

111 

112 url = '/snapshot-instances' 

113 if snapshot_id: 

114 url += '?snapshot_id=%s' % snapshot_id 

115 

116 req = fakes.HTTPRequest.blank(url, version=self.api_version) 

117 req_context = req.environ['manila.context'] 

118 res_dict = self.controller.index(req) 

119 

120 self.assertEqual([expected_snapshot_instance], 

121 res_dict['snapshot_instances']) 

122 self.mock_policy_check.assert_called_once_with( 

123 req_context, self.resource_name, 'index') 

124 

125 def test_list_snapshot_instances_detail(self): 

126 snapshot_instance, expected_snapshot_instance = ( 

127 self._get_fake_snapshot_instance()) 

128 self.mock_object(share_snapshot_instances.db, 

129 'share_snapshot_instance_get_all_with_filters', 

130 mock.Mock(return_value=[snapshot_instance])) 

131 

132 res_dict = self.controller.detail(self.snapshot_instances_req) 

133 

134 self.assertEqual([expected_snapshot_instance], 

135 res_dict['snapshot_instances']) 

136 self.mock_policy_check.assert_called_once_with( 

137 self.admin_context, self.resource_name, 'detail') 

138 

139 def test_list_snapshot_instances_detail_invalid_snapshot(self): 

140 self.mock_object(share_snapshot_instances.db, 

141 'share_snapshot_instance_get_all_with_filters', 

142 mock.Mock(return_value=[])) 

143 

144 req = self.snapshot_instances_req 

145 req.GET['snapshot_id'] = 'FAKE_SNAPSHOT_ID' 

146 

147 res_dict = self.controller.detail(req) 

148 

149 self.assertEqual([], res_dict['snapshot_instances']) 

150 self.mock_policy_check.assert_called_once_with( 

151 self.admin_context, self.resource_name, 'detail') 

152 

153 def test_show(self): 

154 snapshot_instance, expected_snapshot_instance = ( 

155 self._get_fake_snapshot_instance()) 

156 self.mock_object( 

157 share_snapshot_instances.db, 'share_snapshot_instance_get', 

158 mock.Mock(return_value=snapshot_instance)) 

159 

160 res_dict = self.controller.show(self.snapshot_instances_req, 

161 snapshot_instance.get('id')) 

162 

163 self.assertEqual(expected_snapshot_instance, 

164 res_dict['snapshot_instance']) 

165 self.mock_policy_check.assert_called_once_with( 

166 self.admin_context, self.resource_name, 'show') 

167 

168 def test_show_snapshot_instance_not_found(self): 

169 mock__view_builder_call = self.mock_object( 

170 share_snapshot_instances.instance_view.ViewBuilder, 'detail') 

171 fake_exception = exception.ShareSnapshotInstanceNotFound( 

172 instance_id='FAKE_SNAPSHOT_INSTANCE_ID') 

173 self.mock_object(share_snapshot_instances.db, 

174 'share_snapshot_instance_get', 

175 mock.Mock(side_effect=fake_exception)) 

176 

177 self.assertRaises(exc.HTTPNotFound, 

178 self.controller.show, 

179 self.snapshot_instances_req, 

180 'FAKE_SNAPSHOT_INSTANCE_ID') 

181 self.assertFalse(mock__view_builder_call.called) 

182 

183 @ddt.data('index', 'detail', 'show', 'reset_status') 

184 def test_policy_not_authorized(self, method_name): 

185 

186 method = getattr(self.controller, method_name) 

187 if method_name in ('index', 'detail'): 

188 arguments = {} 

189 else: 

190 arguments = { 

191 'id': 'FAKE_SNAPSHOT_ID', 

192 'body': {'FAKE_KEY': 'FAKE_VAL'}, 

193 } 

194 

195 noauthexc = exception.PolicyNotAuthorized(action=method_name) 

196 

197 with mock.patch.object( 

198 policy, 'check_policy', mock.Mock(side_effect=noauthexc)): 

199 

200 self.assertRaises( 

201 exc.HTTPForbidden, method, self.snapshot_instances_req, 

202 **arguments) 

203 

204 @ddt.data('index', 'show', 'detail', 'reset_status') 

205 def test_upsupported_microversion(self, method_name): 

206 unsupported_microversions = ('1.0', '2.18') 

207 method = getattr(self.controller, method_name) 

208 arguments = { 

209 'id': 'FAKE_SNAPSHOT_ID', 

210 } 

211 if method_name in ('index'): 

212 arguments.clear() 

213 

214 for microversion in unsupported_microversions: 

215 req = fakes.HTTPRequest.blank( 

216 '/snapshot-instances', version=microversion) 

217 self.assertRaises(exception.VersionNotFoundForAPIMethod, 

218 method, req, **arguments) 

219 

220 def _reset_status(self, context, instance, req, 

221 valid_code=202, valid_status=None, body=None): 

222 if body is None: 222 ↛ 225line 222 didn't jump to line 225 because the condition on line 222 was always true

223 body = {'reset_status': {'status': constants.STATUS_ERROR}} 

224 

225 req.body = jsonutils.dumps(body).encode("utf-8") 

226 req.environ['manila.context'] = context 

227 

228 with mock.patch.object( 

229 policy, 'check_policy', fakes.mock_fake_admin_check): 

230 resp = req.get_response(fakes.app()) 

231 

232 # validate response code and model status 

233 self.assertEqual(valid_code, resp.status_int) 

234 

235 actual_instance = ( 

236 share_snapshot_instances.db.share_snapshot_instance_get( 

237 context, instance['id'])) 

238 self.assertEqual(valid_status, actual_instance['status']) 

239 

240 @ddt.data(*fakes.fixture_reset_status_with_different_roles) 

241 @ddt.unpack 

242 def test_reset_status_with_different_roles(self, role, valid_code, 

243 valid_status, version): 

244 instance, action_req = self._setup_snapshot_instance_data() 

245 ctxt = self._get_context(role) 

246 self._reset_status(ctxt, instance, action_req, 

247 valid_code=valid_code, 

248 valid_status=valid_status)