Coverage for manila/tests/api/fakes.py: 81%

1# Copyright 2010 OpenStack LLC. 

2# All Rights Reserved. 

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 

16from oslo_service import wsgi 

17from oslo_utils import timeutils 

18from oslo_utils import uuidutils 

19import routes 

20import webob 

21import webob.dec 

22import webob.request 

23 

24from manila.api import common as api_common 

25from manila.api.openstack import api_version_request as api_version 

26from manila.api.openstack import wsgi as os_wsgi 

27from manila.api import urlmap 

28from manila.api.v1 import router as router_v1 

29from manila.api.v2 import router as router_v2 

30from manila.common import constants 

31from manila import context 

32from manila import exception 

33 

34 

35CONTEXT = context.get_admin_context() 

36driver_opts = {} 

37FAKE_UUID = '123e4567-e89b-12d3-a456-426614174000' 

38FAKE_UUIDS = {} 

39host = 'host_name' 

40identifier = '7cf7c200-d3af-4e05-b87e-9167c95dfcad' 

41 

42 

43class Context(object): 

44 pass 

45 

46 

47class FakeRouter(wsgi.Router): 

48 def __init__(self, ext_mgr=None): 

49 pass 

50 

51 @webob.dec.wsgify 

52 def __call__(self, req): 

53 res = webob.Response() 

54 res.status = '200' 

55 res.headers['X-Test-Success'] = 'True' 

56 return res 

57 

58 

59@webob.dec.wsgify 

60def fake_wsgi(self, req): 

61 return self.application 

62 

63 

64class FakeToken(object): 

65 id_count = 0 

66 

67 def __getitem__(self, key): 

68 return getattr(self, key) 

69 

70 def __init__(self, **kwargs): 

71 FakeToken.id_count += 1 

72 self.id = FakeToken.id_count 

73 self.token_hash = None 

74 for k, v in kwargs.items(): 

75 setattr(self, k, v) 

76 

77 

78class FakeRequestContext(context.RequestContext): 

79 def __init__(self, *args, **kwargs): 

80 kwargs['auth_token'] = kwargs.get('auth_token', 'fake_auth_token') 

81 super(FakeRequestContext, self).__init__(*args, **kwargs) 

82 

83 

84class HTTPRequest(os_wsgi.Request): 

85 

86 @classmethod 

87 def blank(cls, *args, **kwargs): 

88 if not kwargs.get('base_url'): 

89 method_url = args[0] 

90 if method_url.startswith('/v2'): 

91 kwargs['base_url'] = 'http://localhost/share/v2' 

92 else: 

93 kwargs['base_url'] = 'http://localhost/share/v1' 

94 use_admin_context = kwargs.pop('use_admin_context', False) 

95 version = kwargs.pop('version', api_version.DEFAULT_API_VERSION) 

96 experimental = kwargs.pop('experimental', False) 

97 out = os_wsgi.Request.blank(*args, **kwargs) 

98 out.environ['manila.context'] = FakeRequestContext( 

99 'fake_user', 

100 'fake', 

101 is_admin=use_admin_context) 

102 out.api_version_request = api_version.APIVersionRequest( 

103 version, experimental=experimental) 

104 return out 

105 

106 

107class TestRouter(wsgi.Router): 

108 def __init__(self, controller): 

109 mapper = routes.Mapper() 

110 mapper.resource("test", "tests", 

111 controller=os_wsgi.Resource(controller)) 

112 super(TestRouter, self).__init__(mapper) 

113 

114 

115class FakeAuthDatabase(object): 

116 data = {} 

117 

118 @staticmethod 

119 def auth_token_get(context, token_hash): 

120 return FakeAuthDatabase.data.get(token_hash, None) 

121 

122 @staticmethod 

123 def auth_token_create(context, token): 

124 fake_token = FakeToken(created_at=timeutils.utcnow(), **token) 

125 FakeAuthDatabase.data[fake_token.token_hash] = fake_token 

126 FakeAuthDatabase.data['id_%i' % fake_token.id] = fake_token 

127 return fake_token 

128 

129 @staticmethod 

130 def auth_token_destroy(context, token_id): 

131 token = FakeAuthDatabase.data.get('id_%i' % token_id) 

132 if token and token.token_hash in FakeAuthDatabase.data: 

133 del FakeAuthDatabase.data[token.token_hash] 

134 del FakeAuthDatabase.data['id_%i' % token_id] 

135 

136 

137class FakeRateLimiter(object): 

138 def __init__(self, application): 

139 self.application = application 

140 

141 @webob.dec.wsgify 

142 def __call__(self, req): 

143 return self.application 

144 

145 

146def get_fake_uuid(token=0): 

147 if token not in FAKE_UUIDS: 

148 FAKE_UUIDS[token] = uuidutils.generate_uuid() 

149 return FAKE_UUIDS[token] 

150 

151 

152def app(): 

153 """API application. 

154 

155 No auth, just let environ['manila.context'] pass through. 

156 """ 

157 mapper = urlmap.URLMap() 

158 mapper['/v1'] = router_v1.APIRouter() 

159 mapper['/v2'] = router_v2.APIRouter() 

160 return mapper 

161 

162 

163fixture_reset_status_with_different_roles_v1 = ( 

164 { 

165 'role': 'admin', 

166 'valid_code': 202, 

167 'valid_status': constants.STATUS_ERROR, 

168 }, 

169 { 

170 'role': 'member', 

171 'valid_code': 403, 

172 'valid_status': constants.STATUS_AVAILABLE, 

173 }, 

174) 

175 

176fixture_reset_status_with_different_roles = ( 

177 { 

178 'role': 'admin', 

179 'valid_code': 202, 

180 'valid_status': constants.STATUS_ERROR, 

181 'version': '2.6', 

182 }, 

183 { 

184 'role': 'admin', 

185 'valid_code': 202, 

186 'valid_status': constants.STATUS_ERROR, 

187 'version': '2.7', 

188 }, 

189 { 

190 'role': 'member', 

191 'valid_code': 403, 

192 'valid_status': constants.STATUS_AVAILABLE, 

193 'version': '2.6', 

194 }, 

195 { 

196 'role': 'member', 

197 'valid_code': 403, 

198 'valid_status': constants.STATUS_AVAILABLE, 

199 'version': '2.7', 

200 }, 

201) 

202 

203 

204fixture_reset_replica_status_with_different_roles = ( 

205 { 

206 'role': 'admin', 

207 'valid_code': 202, 

208 'valid_status': constants.STATUS_ERROR, 

209 'microversion': '2.55' 

210 }, 

211 { 

212 'role': 'admin', 

213 'valid_code': 202, 

214 'valid_status': constants.STATUS_ERROR, 

215 'microversion': '2.56' 

216 }, 

217 { 

218 'role': 'member', 

219 'valid_code': 403, 

220 'valid_status': constants.STATUS_AVAILABLE, 

221 'microversion': '2.55' 

222 }, 

223) 

224 

225 

226fixture_reset_replica_state_with_different_roles = ( 

227 { 

228 'role': 'admin', 

229 'valid_code': 202, 

230 'valid_status': constants.REPLICA_STATE_ACTIVE, 

231 'microversion': '2.55' 

232 }, 

233 { 

234 'role': 'admin', 

235 'valid_code': 202, 

236 'valid_status': constants.REPLICA_STATE_OUT_OF_SYNC, 

237 'microversion': '2.56' 

238 }, 

239 { 

240 'role': 'admin', 

241 'valid_code': 202, 

242 'valid_status': constants.REPLICA_STATE_IN_SYNC, 

243 'microversion': '2.55' 

244 }, 

245 { 

246 'role': 'admin', 

247 'valid_code': 202, 

248 'valid_status': constants.STATUS_ERROR, 

249 'microversion': '2.56' 

250 }, 

251 { 

252 'role': 'member', 

253 'valid_code': 403, 

254 'valid_status': constants.REPLICA_STATE_IN_SYNC, 

255 'microversion': '2.55' 

256 }, 

257) 

258 

259 

260fixture_force_delete_with_different_roles = ( 

261 {'role': 'admin', 'resp_code': 202, 'version': '2.6'}, 

262 {'role': 'admin', 'resp_code': 202, 'version': '2.7'}, 

263 {'role': 'member', 'resp_code': 403, 'version': '2.6'}, 

264 {'role': 'member', 'resp_code': 403, 'version': '2.7'}, 

265) 

266 

267 

268fixture_invalid_reset_status_body = ( 

269 {'os-reset_status': {'x-status': 'bad'}}, 

270 {'os-reset_status': {'status': 'invalid'}} 

271) 

272 

273 

274fixture_valid_reset_status_body = ( 

275 ({'os-reset_status': {'status': 'creating'}}, '2.6'), 

276 ({'os-reset_status': {'status': 'available'}}, '2.6'), 

277 ({'os-reset_status': {'status': 'deleting'}}, '2.6'), 

278 ({'os-reset_status': {'status': 'error_deleting'}}, '2.6'), 

279 ({'os-reset_status': {'status': 'error'}}, '2.6'), 

280 ({'os-reset_status': {'status': 'migrating'}}, '2.6'), 

281 ({'os-reset_status': {'status': 'migrating_to'}}, '2.6'), 

282 ({'reset_status': {'status': 'creating'}}, '2.7'), 

283 ({'reset_status': {'status': 'available'}}, '2.7'), 

284 ({'reset_status': {'status': 'deleting'}}, '2.7'), 

285 ({'reset_status': {'status': 'error_deleting'}}, '2.7'), 

286 ({'reset_status': {'status': 'error'}}, '2.7'), 

287 ({'reset_status': {'status': 'migrating'}}, '2.7'), 

288 ({'reset_status': {'status': 'migrating_to'}}, '2.7'), 

289) 

290 

291 

292def mock_fake_admin_check(context, resource_name, action, *args, **kwargs): 

293 if context.is_admin: 

294 return 

295 else: 

296 raise exception.PolicyNotAuthorized(action=action) 

297 

298 

299class FakeResourceViewBuilder(api_common.ViewBuilder): 

300 

301 _collection_name = 'fake_resource' 

302 _detail_version_modifiers = [ 

303 "add_field_xyzzy", 

304 "add_field_spoon_for_admins", 

305 "remove_field_foo", 

306 ] 

307 

308 def view(self, req, resource): 

309 

310 keys = ('id', 'foo', 'fred', 'alice') 

311 resource_dict = {key: resource.get(key) for key in keys} 

312 

313 self.update_versioned_resource_dict(req, resource_dict, resource) 

314 

315 return resource_dict 

316 

317 @api_common.ViewBuilder.versioned_method("1.41") 

318 def add_field_xyzzy(self, context, resource_dict, resource): 

319 resource_dict['xyzzy'] = resource.get('xyzzy') 

320 

321 @api_common.ViewBuilder.versioned_method("1.6") 

322 def add_field_spoon_for_admins(self, context, resource_dict, resource): 

323 if context.is_admin: 

324 resource_dict['spoon'] = resource.get('spoon') 

325 

326 @api_common.ViewBuilder.versioned_method("3.14") 

327 def remove_field_foo(self, context, resource_dict, resource): 

328 resource_dict.pop('foo', None)