Coverage for manila/tests/api/test_versions.py: 98%

1# Copyright 2015 Clinton Knight 

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 unittest import mock 

17 

18import ddt 

19from oslo_serialization import jsonutils 

20from oslo_utils import encodeutils 

21 

22from manila.api.openstack import api_version_request 

23from manila.api.openstack import wsgi 

24from manila.api.v1 import router 

25from manila.api import versions 

26from manila import test 

27from manila.tests.api import fakes 

28 

29 

30version_header_name = 'X-OpenStack-Manila-API-Version' 

31experimental_header_name = 'X-OpenStack-Manila-API-Experimental' 

32 

33 

34@ddt.ddt 

35class VersionsControllerTestCase(test.TestCase): 

36 

37 def setUp(self): 

38 super(VersionsControllerTestCase, self).setUp() 

39 self.wsgi_apps = (versions.VersionsRouter(), router.APIRouter()) 

40 

41 @ddt.data('1.0', '1.1', '2.0', '3.0') 

42 def test_versions_root(self, version): 

43 req = fakes.HTTPRequest.blank('/', base_url='http://localhost') 

44 req.method = 'GET' 

45 req.content_type = 'application/json' 

46 req.headers = {version_header_name: version} 

47 

48 response = req.get_response(versions.VersionsRouter()) 

49 self.assertEqual(300, response.status_int) 

50 body = jsonutils.loads(response.body) 

51 version_list = body['versions'] 

52 

53 ids = [v['id'] for v in version_list] 

54 self.assertEqual({'v1.0', 'v2.0'}, set(ids)) 

55 self.assertNotIn(version_header_name, response.headers) 

56 self.assertNotIn('Vary', response.headers) 

57 

58 v1 = [v for v in version_list if v['id'] == 'v1.0'][0] 

59 self.assertEqual('', v1.get('min_version')) 

60 self.assertEqual('', v1.get('version')) 

61 self.assertEqual('DEPRECATED', v1.get('status')) 

62 

63 v2 = [v for v in version_list if v['id'] == 'v2.0'][0] 

64 self.assertEqual(api_version_request._MIN_API_VERSION, 

65 v2.get('min_version')) 

66 self.assertEqual(api_version_request._MAX_API_VERSION, 

67 v2.get('version')) 

68 self.assertEqual('CURRENT', v2.get('status')) 

69 

70 @ddt.data('1.0', 

71 '1.1', 

72 api_version_request._MIN_API_VERSION, 

73 api_version_request._MAX_API_VERSION) 

74 def test_versions_v1(self, version): 

75 req = fakes.HTTPRequest.blank('/', base_url='http://localhost/v1') 

76 req.method = 'GET' 

77 req.content_type = 'application/json' 

78 req.headers = {version_header_name: version} 

79 

80 response = req.get_response(router.APIRouter()) 

81 self.assertEqual(200, response.status_int) 

82 body = jsonutils.loads(response.body) 

83 version_list = body['versions'] 

84 

85 ids = [v['id'] for v in version_list] 

86 self.assertEqual({'v1.0'}, set(ids)) 

87 self.assertEqual('1.0', response.headers[version_header_name]) 

88 self.assertEqual(version_header_name, response.headers['Vary']) 

89 self.assertEqual('', version_list[0].get('min_version')) 

90 self.assertEqual('', version_list[0].get('version')) 

91 self.assertEqual('DEPRECATED', version_list[0].get('status')) 

92 

93 @ddt.data(api_version_request._MIN_API_VERSION, 

94 api_version_request._MAX_API_VERSION) 

95 def test_versions_v2(self, version): 

96 req = fakes.HTTPRequest.blank('/', base_url='http://localhost/v2') 

97 req.method = 'GET' 

98 req.content_type = 'application/json' 

99 req.headers = {version_header_name: version} 

100 

101 response = req.get_response(router.APIRouter()) 

102 self.assertEqual(200, response.status_int) 

103 body = jsonutils.loads(response.body) 

104 version_list = body['versions'] 

105 

106 ids = [v['id'] for v in version_list] 

107 self.assertEqual({'v2.0'}, set(ids)) 

108 self.assertEqual(version, response.headers[version_header_name]) 

109 self.assertEqual(version_header_name, response.headers['Vary']) 

110 

111 v2 = [v for v in version_list if v['id'] == 'v2.0'][0] 

112 self.assertEqual(api_version_request._MIN_API_VERSION, 

113 v2.get('min_version')) 

114 self.assertEqual(api_version_request._MAX_API_VERSION, 

115 v2.get('version')) 

116 

117 def test_versions_version_invalid(self): 

118 req = fakes.HTTPRequest.blank('/', base_url='http://localhost/v2') 

119 req.method = 'GET' 

120 req.content_type = 'application/json' 

121 req.headers = {version_header_name: '2.0.1'} 

122 

123 for app in self.wsgi_apps: 

124 response = req.get_response(app) 

125 

126 self.assertEqual(400, response.status_int) 

127 

128 def test_versions_version_not_found(self): 

129 api_version_request_3_0 = api_version_request.APIVersionRequest('3.0') 

130 self.mock_object(api_version_request, 

131 'max_api_version', 

132 mock.Mock(return_value=api_version_request_3_0)) 

133 

134 class Controller(wsgi.Controller): 

135 @wsgi.Controller.api_version('2.0', '2.0') 

136 def index(self, req): 

137 return 'off' 

138 

139 req = fakes.HTTPRequest.blank('/tests', base_url='http://localhost/v2') 

140 req.headers = {version_header_name: '2.5'} 

141 app = fakes.TestRouter(Controller()) 

142 

143 response = req.get_response(app) 

144 

145 self.assertEqual(404, response.status_int) 

146 

147 def test_versions_version_not_acceptable(self): 

148 req = fakes.HTTPRequest.blank('/', base_url='http://localhost/v2') 

149 req.method = 'GET' 

150 req.content_type = 'application/json' 

151 req.headers = {version_header_name: '3.0'} 

152 

153 response = req.get_response(router.APIRouter()) 

154 

155 self.assertEqual(406, response.status_int) 

156 self.assertEqual('3.0', response.headers[version_header_name]) 

157 self.assertEqual(version_header_name, response.headers['Vary']) 

158 

159 @ddt.data(['2.5', 200], ['2.55', 404]) 

160 @ddt.unpack 

161 def test_req_version_matches(self, version, HTTP_ret): 

162 version_request = api_version_request.APIVersionRequest(version) 

163 self.mock_object(api_version_request, 

164 'max_api_version', 

165 mock.Mock(return_value=version_request)) 

166 

167 class Controller(wsgi.Controller): 

168 

169 @wsgi.Controller.api_version('2.0', '2.6') 

170 def index(self, req): 

171 return 'off' 

172 

173 req = fakes.HTTPRequest.blank('/tests', base_url='http://localhost/v2') 

174 req.headers = {version_header_name: version} 

175 app = fakes.TestRouter(Controller()) 

176 

177 response = req.get_response(app) 

178 

179 if HTTP_ret == 200: 

180 self.assertEqual(b'off', response.body) 

181 elif HTTP_ret == 404: 181 ↛ 183line 181 didn't jump to line 183 because the condition on line 181 was always true

182 self.assertNotEqual(b'off', response.body) 

183 self.assertEqual(HTTP_ret, response.status_int) 

184 

185 @ddt.data(['2.5', 'older'], ['2.37', 'newer']) 

186 @ddt.unpack 

187 def test_req_version_matches_with_if(self, version, ret_val): 

188 version_request = api_version_request.APIVersionRequest(version) 

189 self.mock_object(api_version_request, 

190 'max_api_version', 

191 mock.Mock(return_value=version_request)) 

192 

193 class Controller(wsgi.Controller): 

194 

195 def index(self, req): 

196 req_version = req.api_version_request 

197 if req_version.matches('2.1', '2.8'): 

198 return 'older' 

199 if req_version.matches('2.9', '2.88'): 199 ↛ exitline 199 didn't return from function 'index' because the condition on line 199 was always true

200 return 'newer' 

201 

202 req = fakes.HTTPRequest.blank('/tests', base_url='http://localhost/v2') 

203 req.headers = {version_header_name: version} 

204 app = fakes.TestRouter(Controller()) 

205 

206 response = req.get_response(app) 

207 

208 resp = encodeutils.safe_decode(response.body, incoming='utf-8') 

209 self.assertEqual(ret_val, resp) 

210 self.assertEqual(200, response.status_int) 

211 

212 @ddt.data(['2.5', 'older'], ['2.37', 'newer']) 

213 @ddt.unpack 

214 def test_req_version_matches_with_None(self, version, ret_val): 

215 version_request = api_version_request.APIVersionRequest(version) 

216 self.mock_object(api_version_request, 

217 'max_api_version', 

218 mock.Mock(return_value=version_request)) 

219 

220 class Controller(wsgi.Controller): 

221 

222 def index(self, req): 

223 req_version = req.api_version_request 

224 if req_version.matches(None, '2.8'): 

225 return 'older' 

226 if req_version.matches('2.9', None): 226 ↛ exitline 226 didn't return from function 'index' because the condition on line 226 was always true

227 return 'newer' 

228 

229 req = fakes.HTTPRequest.blank('/tests', base_url='http://localhost/v2') 

230 req.headers = {version_header_name: version} 

231 app = fakes.TestRouter(Controller()) 

232 

233 response = req.get_response(app) 

234 

235 resp = encodeutils.safe_decode(response.body, incoming='utf-8') 

236 self.assertEqual(ret_val, resp) 

237 self.assertEqual(200, response.status_int) 

238 

239 def test_req_version_matches_with_None_None(self): 

240 version_request = api_version_request.APIVersionRequest('2.39') 

241 self.mock_object(api_version_request, 

242 'max_api_version', 

243 mock.Mock(return_value=version_request)) 

244 

245 class Controller(wsgi.Controller): 

246 

247 def index(self, req): 

248 req_version = req.api_version_request 

249 # This case is artificial, and will return True 

250 if req_version.matches(None, None): 250 ↛ exitline 250 didn't return from function 'index' because the condition on line 250 was always true

251 return "Pass" 

252 

253 req = fakes.HTTPRequest.blank('/tests', base_url='http://localhost/v2') 

254 req.headers = {version_header_name: '2.39'} 

255 app = fakes.TestRouter(Controller()) 

256 

257 response = req.get_response(app) 

258 

259 resp = encodeutils.safe_decode(response.body, incoming='utf-8') 

260 self.assertEqual("Pass", resp) 

261 self.assertEqual(200, response.status_int) 

262 

263 

264@ddt.ddt 

265class ExperimentalAPITestCase(test.TestCase): 

266 

267 class Controller(wsgi.Controller): 

268 @wsgi.Controller.api_version('2.0', '2.0') 

269 def index(self, req): 

270 return {'fake_key': 'fake_value'} 

271 

272 @wsgi.Controller.api_version('2.1', '2.1', experimental=True) # noqa 

273 def index(self, req): # pylint: disable=function-redefined # noqa F811 

274 return {'fake_key': 'fake_value'} 

275 

276 def setUp(self): 

277 super(ExperimentalAPITestCase, self).setUp() 

278 self.app = fakes.TestRouter(ExperimentalAPITestCase.Controller()) 

279 self.req = fakes.HTTPRequest.blank('/tests', 

280 base_url='http://localhost/v2') 

281 

282 @ddt.data(True, False) 

283 def test_stable_api_always_called(self, experimental): 

284 

285 self.req.headers = {version_header_name: '2.0'} 

286 if experimental: 

287 self.req.headers[experimental_header_name] = experimental 

288 response = self.req.get_response(self.app) 

289 

290 self.assertEqual(200, response.status_int) 

291 self.assertEqual('2.0', response.headers[version_header_name]) 

292 

293 if experimental: 

294 self.assertEqual('%s' % experimental, 

295 response.headers.get(experimental_header_name)) 

296 else: 

297 self.assertNotIn(experimental_header_name, response.headers) 

298 

299 def test_experimental_api_called_when_requested(self): 

300 

301 self.req.headers = { 

302 version_header_name: '2.1', 

303 experimental_header_name: 'True', 

304 } 

305 response = self.req.get_response(self.app) 

306 

307 self.assertEqual(200, response.status_int) 

308 self.assertEqual('2.1', response.headers[version_header_name]) 

309 self.assertTrue(response.headers.get(experimental_header_name)) 

310 

311 def test_experimental_api_not_called_when_not_requested(self): 

312 

313 self.req.headers = {version_header_name: '2.1'} 

314 response = self.req.get_response(self.app) 

315 

316 self.assertEqual(404, response.status_int) 

317 self.assertNotIn(experimental_header_name, response.headers) 

318 

319 def test_experimental_header_returned_in_exception(self): 

320 

321 api_version_request_3_0 = api_version_request.APIVersionRequest('3.0') 

322 self.mock_object(api_version_request, 

323 'max_api_version', 

324 mock.Mock(return_value=api_version_request_3_0)) 

325 

326 self.req.headers = { 

327 version_header_name: '2.2', 

328 experimental_header_name: 'True', 

329 } 

330 response = self.req.get_response(self.app) 

331 

332 self.assertEqual(404, response.status_int) 

333 self.assertTrue(response.headers.get(experimental_header_name))