summaryrefslogtreecommitdiff
path: root/novaclient/tests/unit/v2/test_auth.py
diff options
context:
space:
mode:
Diffstat (limited to 'novaclient/tests/unit/v2/test_auth.py')
-rw-r--r--novaclient/tests/unit/v2/test_auth.py386
1 files changed, 386 insertions, 0 deletions
diff --git a/novaclient/tests/unit/v2/test_auth.py b/novaclient/tests/unit/v2/test_auth.py
new file mode 100644
index 0000000..95e96de
--- /dev/null
+++ b/novaclient/tests/unit/v2/test_auth.py
@@ -0,0 +1,386 @@
1#
2# Licensed under the Apache License, Version 2.0 (the "License"); you may
3# not use this file except in compliance with the License. You may obtain
4# a copy of the License at
5#
6# http://www.apache.org/licenses/LICENSE-2.0
7#
8# Unless required by applicable law or agreed to in writing, software
9# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
10# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
11# License for the specific language governing permissions and limitations
12# under the License.
13
14import copy
15import json
16
17from keystoneclient import fixture
18import mock
19import requests
20
21from novaclient import exceptions
22from novaclient.tests.unit import utils
23from novaclient.v2 import client
24
25
26class AuthenticateAgainstKeystoneTests(utils.TestCase):
27
28 def get_token(self, **kwargs):
29 resp = fixture.V2Token(**kwargs)
30 resp.set_scope()
31
32 s = resp.add_service('compute')
33 s.add_endpoint('http://localhost:8774/v1.1', region='RegionOne')
34
35 return resp
36
37 def test_authenticate_success(self):
38 cs = client.Client("username", "password", "project_id",
39 utils.AUTH_URL_V2, service_type='compute')
40 resp = self.get_token()
41
42 auth_response = utils.TestResponse({
43 "status_code": 200,
44 "text": json.dumps(resp),
45 })
46
47 mock_request = mock.Mock(return_value=(auth_response))
48
49 @mock.patch.object(requests, "request", mock_request)
50 def test_auth_call():
51 cs.client.authenticate()
52 headers = {
53 'User-Agent': cs.client.USER_AGENT,
54 'Content-Type': 'application/json',
55 'Accept': 'application/json',
56 }
57 body = {
58 'auth': {
59 'passwordCredentials': {
60 'username': cs.client.user,
61 'password': cs.client.password,
62 },
63 'tenantName': cs.client.projectid,
64 },
65 }
66
67 token_url = cs.client.auth_url + "/tokens"
68 mock_request.assert_called_with(
69 "POST",
70 token_url,
71 headers=headers,
72 data=json.dumps(body),
73 allow_redirects=True,
74 **self.TEST_REQUEST_BASE)
75
76 endpoints = resp["access"]["serviceCatalog"][0]['endpoints']
77 public_url = endpoints[0]["publicURL"].rstrip('/')
78 self.assertEqual(cs.client.management_url, public_url)
79 token_id = resp["access"]["token"]["id"]
80 self.assertEqual(cs.client.auth_token, token_id)
81
82 test_auth_call()
83
84 def test_authenticate_failure(self):
85 cs = client.Client("username", "password", "project_id",
86 utils.AUTH_URL_V2)
87 resp = {"unauthorized": {"message": "Unauthorized", "code": "401"}}
88 auth_response = utils.TestResponse({
89 "status_code": 401,
90 "text": json.dumps(resp),
91 })
92
93 mock_request = mock.Mock(return_value=(auth_response))
94
95 @mock.patch.object(requests.Session, "request", mock_request)
96 def test_auth_call():
97 self.assertRaises(exceptions.Unauthorized, cs.client.authenticate)
98
99 test_auth_call()
100
101 def test_v1_auth_redirect(self):
102 cs = client.Client("username", "password", "project_id",
103 utils.AUTH_URL_V1, service_type='compute')
104 dict_correct_response = self.get_token()
105 correct_response = json.dumps(dict_correct_response)
106 dict_responses = [
107 {"headers": {'location': 'http://127.0.0.1:5001'},
108 "status_code": 305,
109 "text": "Use proxy"},
110 # Configured on admin port, nova redirects to v2.0 port.
111 # When trying to connect on it, keystone auth succeed by v1.0
112 # protocol (through headers) but tokens are being returned in
113 # body (looks like keystone bug). Leaved for compatibility.
114 {"headers": {},
115 "status_code": 200,
116 "text": correct_response},
117 {"headers": {},
118 "status_code": 200,
119 "text": correct_response}
120 ]
121
122 responses = [(utils.TestResponse(resp)) for resp in dict_responses]
123
124 def side_effect(*args, **kwargs):
125 return responses.pop(0)
126
127 mock_request = mock.Mock(side_effect=side_effect)
128
129 @mock.patch.object(requests, "request", mock_request)
130 def test_auth_call():
131 cs.client.authenticate()
132 headers = {
133 'User-Agent': cs.client.USER_AGENT,
134 'Content-Type': 'application/json',
135 'Accept': 'application/json',
136 }
137 body = {
138 'auth': {
139 'passwordCredentials': {
140 'username': cs.client.user,
141 'password': cs.client.password,
142 },
143 'tenantName': cs.client.projectid,
144 },
145 }
146
147 token_url = cs.client.auth_url + "/tokens"
148 kwargs = copy.copy(self.TEST_REQUEST_BASE)
149 kwargs['headers'] = headers
150 kwargs['data'] = json.dumps(body)
151 mock_request.assert_called_with(
152 "POST",
153 token_url,
154 allow_redirects=True,
155 **kwargs)
156
157 resp = dict_correct_response
158 endpoints = resp["access"]["serviceCatalog"][0]['endpoints']
159 public_url = endpoints[0]["publicURL"].rstrip('/')
160 self.assertEqual(cs.client.management_url, public_url)
161 token_id = resp["access"]["token"]["id"]
162 self.assertEqual(cs.client.auth_token, token_id)
163
164 test_auth_call()
165
166 def test_v2_auth_redirect(self):
167 cs = client.Client("username", "password", "project_id",
168 utils.AUTH_URL_V2, service_type='compute')
169 dict_correct_response = self.get_token()
170 correct_response = json.dumps(dict_correct_response)
171 dict_responses = [
172 {"headers": {'location': 'http://127.0.0.1:5001'},
173 "status_code": 305,
174 "text": "Use proxy"},
175 # Configured on admin port, nova redirects to v2.0 port.
176 # When trying to connect on it, keystone auth succeed by v1.0
177 # protocol (through headers) but tokens are being returned in
178 # body (looks like keystone bug). Leaved for compatibility.
179 {"headers": {},
180 "status_code": 200,
181 "text": correct_response},
182 {"headers": {},
183 "status_code": 200,
184 "text": correct_response}
185 ]
186
187 responses = [(utils.TestResponse(resp)) for resp in dict_responses]
188
189 def side_effect(*args, **kwargs):
190 return responses.pop(0)
191
192 mock_request = mock.Mock(side_effect=side_effect)
193
194 @mock.patch.object(requests, "request", mock_request)
195 def test_auth_call():
196 cs.client.authenticate()
197 headers = {
198 'User-Agent': cs.client.USER_AGENT,
199 'Content-Type': 'application/json',
200 'Accept': 'application/json',
201 }
202 body = {
203 'auth': {
204 'passwordCredentials': {
205 'username': cs.client.user,
206 'password': cs.client.password,
207 },
208 'tenantName': cs.client.projectid,
209 },
210 }
211
212 token_url = cs.client.auth_url + "/tokens"
213 kwargs = copy.copy(self.TEST_REQUEST_BASE)
214 kwargs['headers'] = headers
215 kwargs['data'] = json.dumps(body)
216 mock_request.assert_called_with(
217 "POST",
218 token_url,
219 allow_redirects=True,
220 **kwargs)
221
222 resp = dict_correct_response
223 endpoints = resp["access"]["serviceCatalog"][0]['endpoints']
224 public_url = endpoints[0]["publicURL"].rstrip('/')
225 self.assertEqual(cs.client.management_url, public_url)
226 token_id = resp["access"]["token"]["id"]
227 self.assertEqual(cs.client.auth_token, token_id)
228
229 test_auth_call()
230
231 def test_ambiguous_endpoints(self):
232 cs = client.Client("username", "password", "project_id",
233 utils.AUTH_URL_V2, service_type='compute')
234 resp = self.get_token()
235
236 # duplicate existing service
237 s = resp.add_service('compute')
238 s.add_endpoint('http://localhost:8774/v1.1', region='RegionOne')
239
240 auth_response = utils.TestResponse({
241 "status_code": 200,
242 "text": json.dumps(resp),
243 })
244
245 mock_request = mock.Mock(return_value=(auth_response))
246
247 @mock.patch.object(requests.Session, "request", mock_request)
248 def test_auth_call():
249 self.assertRaises(exceptions.AmbiguousEndpoints,
250 cs.client.authenticate)
251
252 test_auth_call()
253
254 def test_authenticate_with_token_success(self):
255 cs = client.Client("username", None, "project_id",
256 utils.AUTH_URL_V2, service_type='compute')
257 cs.client.auth_token = "FAKE_ID"
258 resp = self.get_token(token_id="FAKE_ID")
259 auth_response = utils.TestResponse({
260 "status_code": 200,
261 "text": json.dumps(resp),
262 })
263
264 mock_request = mock.Mock(return_value=(auth_response))
265
266 with mock.patch.object(requests, "request", mock_request):
267 cs.client.authenticate()
268 headers = {
269 'User-Agent': cs.client.USER_AGENT,
270 'Content-Type': 'application/json',
271 'Accept': 'application/json',
272 }
273 body = {
274 'auth': {
275 'token': {
276 'id': cs.client.auth_token,
277 },
278 'tenantName': cs.client.projectid,
279 },
280 }
281
282 token_url = cs.client.auth_url + "/tokens"
283 mock_request.assert_called_with(
284 "POST",
285 token_url,
286 headers=headers,
287 data=json.dumps(body),
288 allow_redirects=True,
289 **self.TEST_REQUEST_BASE)
290
291 endpoints = resp["access"]["serviceCatalog"][0]['endpoints']
292 public_url = endpoints[0]["publicURL"].rstrip('/')
293 self.assertEqual(cs.client.management_url, public_url)
294 token_id = resp["access"]["token"]["id"]
295 self.assertEqual(cs.client.auth_token, token_id)
296
297 def test_authenticate_with_token_failure(self):
298 cs = client.Client("username", None, "project_id", utils.AUTH_URL_V2)
299 cs.client.auth_token = "FAKE_ID"
300 resp = {"unauthorized": {"message": "Unauthorized", "code": "401"}}
301 auth_response = utils.TestResponse({
302 "status_code": 401,
303 "text": json.dumps(resp),
304 })
305
306 mock_request = mock.Mock(return_value=(auth_response))
307
308 with mock.patch.object(requests.Session, "request", mock_request):
309 self.assertRaises(exceptions.Unauthorized, cs.client.authenticate)
310
311
312class AuthenticationTests(utils.TestCase):
313 def test_authenticate_success(self):
314 cs = client.Client("username", "password",
315 "project_id", utils.AUTH_URL)
316 management_url = 'https://localhost/v1.1/443470'
317 auth_response = utils.TestResponse({
318 'status_code': 204,
319 'headers': {
320 'x-server-management-url': management_url,
321 'x-auth-token': '1b751d74-de0c-46ae-84f0-915744b582d1',
322 },
323 })
324 mock_request = mock.Mock(return_value=(auth_response))
325
326 @mock.patch.object(requests, "request", mock_request)
327 def test_auth_call():
328 cs.client.authenticate()
329 headers = {
330 'Accept': 'application/json',
331 'X-Auth-User': 'username',
332 'X-Auth-Key': 'password',
333 'X-Auth-Project-Id': 'project_id',
334 'User-Agent': cs.client.USER_AGENT
335 }
336 mock_request.assert_called_with(
337 "GET",
338 cs.client.auth_url,
339 headers=headers,
340 **self.TEST_REQUEST_BASE)
341
342 self.assertEqual(cs.client.management_url,
343 auth_response.headers['x-server-management-url'])
344 self.assertEqual(cs.client.auth_token,
345 auth_response.headers['x-auth-token'])
346
347 test_auth_call()
348
349 def test_authenticate_failure(self):
350 cs = client.Client("username", "password",
351 "project_id", utils.AUTH_URL)
352 auth_response = utils.TestResponse({'status_code': 401})
353 mock_request = mock.Mock(return_value=(auth_response))
354
355 @mock.patch.object(requests, "request", mock_request)
356 def test_auth_call():
357 self.assertRaises(exceptions.Unauthorized, cs.client.authenticate)
358
359 test_auth_call()
360
361 def test_auth_automatic(self):
362 cs = client.Client("username", "password",
363 "project_id", utils.AUTH_URL)
364 http_client = cs.client
365 http_client.management_url = ''
366 mock_request = mock.Mock(return_value=(None, None))
367
368 @mock.patch.object(http_client, 'request', mock_request)
369 @mock.patch.object(http_client, 'authenticate')
370 def test_auth_call(m):
371 http_client.get('/')
372 self.assertTrue(m.called)
373 self.assertTrue(mock_request.called)
374
375 test_auth_call()
376
377 def test_auth_manual(self):
378 cs = client.Client("username", "password",
379 "project_id", utils.AUTH_URL)
380
381 @mock.patch.object(cs.client, 'authenticate')
382 def test_auth_call(m):
383 cs.authenticate()
384 self.assertTrue(m.called)
385
386 test_auth_call()