1
# vim: tabstop=5 shiftwidth=4 softtabstop=4
3
# Copyright 2010-2011 OpenStack LLC.
6
# Licensed under the Apache License, Version 2.0 (the "License"); you may
7
# not use this file except in compliance with the License. You may obtain
8
# a copy of the License at
10
# http://www.apache.org/licenses/LICENSE-2.0
12
# Unless required by applicable law or agreed to in writing, software
13
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15
# License for the specific language governing permissions and limitations
21
from xml.dom import minidom
27
from nova import exception
28
from nova import flags
30
from nova.tests.api.openstack import fakes
36
FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
38
FAKE_NETWORKS = [('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12'),
39
('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', '10.0.2.12')]
41
DUPLICATE_NETWORKS = [('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12'),
42
('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12')]
44
INVALID_NETWORKS = [('invalid', 'invalid-ip-address')]
49
"display_name": "test_server",
51
"user_id": 'fake_user_id',
52
"tenant_id": 'fake_tenant_id',
53
"created_at": datetime.datetime(2010, 10, 10, 12, 0, 0),
54
"updated_at": datetime.datetime(2010, 11, 11, 11, 0, 0),
55
"security_groups": [{"id": 1, "name": "test"}],
57
"image_ref": 'http://foo.com/123',
59
"instance_type": {"flavorid": '124'},
63
def return_server_by_id(context, id, session=None):
68
def return_security_group_non_existing(context, project_id, group_name):
69
raise exception.SecurityGroupNotFoundForProject(project_id=project_id,
70
security_group_id=group_name)
73
def return_security_group_get_by_name(context, project_id, group_name):
74
return {'id': 1, 'name': group_name}
77
def return_security_group_get(context, security_group_id, session):
78
return {'id': security_group_id}
81
def return_instance_add_security_group(context, instance_id,
86
class CreateserverextTest(test.TestCase):
89
super(CreateserverextTest, self).setUp()
92
super(CreateserverextTest, self).tearDown()
94
def _make_stub_method(self, canned_return):
95
def stub_method(*args, **kwargs):
99
def _setup_mock_compute_api(self):
101
class MockComputeAPI(nova.compute.API):
104
self.injected_files = None
106
self.user_data = None
109
def create(self, *args, **kwargs):
110
if 'injected_files' in kwargs:
111
self.injected_files = kwargs['injected_files']
113
self.injected_files = None
115
if 'requested_networks' in kwargs:
116
self.networks = kwargs['requested_networks']
120
if 'user_data' in kwargs:
121
self.user_data = kwargs['user_data']
125
return ([{'id': '1234', 'display_name': 'fakeinstance',
128
'project_id': 'fake',
132
'progress': 0}], resv_id)
134
def set_admin_password(self, *args, **kwargs):
137
compute_api = MockComputeAPI()
138
self.stubs.Set(nova.compute, 'API',
139
self._make_stub_method(compute_api))
142
def _setup_mock_network_api(self):
143
fakes.stub_out_nw_api(self.stubs)
145
def _create_security_group_request_dict(self, security_groups):
147
server['name'] = 'new-server-test'
148
server['imageRef'] = 'cedef40a-ed67-4d10-800e-17455edce175'
149
server['flavorRef'] = 1
150
if security_groups is not None:
152
for name in security_groups:
153
sg_list.append({'name': name})
154
server['security_groups'] = sg_list
155
return {'server': server}
157
def _create_networks_request_dict(self, networks):
159
server['name'] = 'new-server-test'
160
server['imageRef'] = 'cedef40a-ed67-4d10-800e-17455edce175'
161
server['flavorRef'] = 1
162
if networks is not None:
164
for uuid, fixed_ip in networks:
165
network_list.append({'uuid': uuid, 'fixed_ip': fixed_ip})
166
server['networks'] = network_list
167
return {'server': server}
169
def _create_user_data_request_dict(self, user_data):
171
server['name'] = 'new-server-test'
172
server['imageRef'] = 'cedef40a-ed67-4d10-800e-17455edce175'
173
server['flavorRef'] = 1
174
server['user_data'] = user_data
175
return {'server': server}
177
def _get_create_request_json(self, body_dict):
178
req = webob.Request.blank('/v2/123/os-create-server-ext')
179
req.headers['Content-Type'] = 'application/json'
181
req.body = json.dumps(body_dict)
184
def _run_create_instance_with_mock_compute_api(self, request):
185
compute_api = self._setup_mock_compute_api()
186
self._setup_mock_network_api()
187
response = request.get_response(fakes.wsgi_app())
188
return compute_api, response
190
def _format_xml_request_body(self, body_dict):
191
server = body_dict['server']
194
'<?xml version="1.0" encoding="UTF-8"?>',
195
'<server xmlns="http://docs.rackspacecloud.com/servers/api/v1.1"',
196
' name="%s" imageRef="%s" flavorRef="%s">' % (
197
server['name'], server['imageRef'], server['flavorRef'])])
198
if 'metadata' in server:
199
metadata = server['metadata']
200
body_parts.append('<metadata>')
201
for item in metadata.iteritems():
202
body_parts.append('<meta key="%s">%s</meta>' % item)
203
body_parts.append('</metadata>')
204
if 'personality' in server:
205
personalities = server['personality']
206
body_parts.append('<personality>')
207
for file in personalities:
208
item = (file['path'], file['contents'])
209
body_parts.append('<file path="%s">%s</file>' % item)
210
body_parts.append('</personality>')
211
if 'networks' in server:
212
networks = server['networks']
213
body_parts.append('<networks>')
214
for network in networks:
215
item = (network['uuid'], network['fixed_ip'])
216
body_parts.append('<network uuid="%s" fixed_ip="%s"></network>'
218
body_parts.append('</networks>')
219
body_parts.append('</server>')
220
return ''.join(body_parts)
222
def _get_create_request_xml(self, body_dict):
223
req = webob.Request.blank('/v2/123/os-create-server-ext')
224
req.content_type = 'application/xml'
225
req.accept = 'application/xml'
227
req.body = self._format_xml_request_body(body_dict)
230
def _create_instance_with_networks_json(self, networks):
231
body_dict = self._create_networks_request_dict(networks)
232
request = self._get_create_request_json(body_dict)
233
compute_api, response = \
234
self._run_create_instance_with_mock_compute_api(request)
235
return request, response, compute_api.networks
237
def _create_instance_with_user_data_json(self, networks):
238
body_dict = self._create_user_data_request_dict(networks)
239
request = self._get_create_request_json(body_dict)
240
compute_api, response = \
241
self._run_create_instance_with_mock_compute_api(request)
242
return request, response, compute_api.user_data
244
def _create_instance_with_networks_xml(self, networks):
245
body_dict = self._create_networks_request_dict(networks)
246
request = self._get_create_request_xml(body_dict)
247
compute_api, response = \
248
self._run_create_instance_with_mock_compute_api(request)
249
return request, response, compute_api.networks
251
def test_create_instance_with_no_networks(self):
252
request, response, networks = \
253
self._create_instance_with_networks_json(networks=None)
254
self.assertEquals(response.status_int, 202)
255
self.assertEquals(networks, None)
257
def test_create_instance_with_no_networks_xml(self):
258
request, response, networks = \
259
self._create_instance_with_networks_xml(networks=None)
260
self.assertEquals(response.status_int, 202)
261
self.assertEquals(networks, None)
263
def test_create_instance_with_one_network(self):
264
request, response, networks = \
265
self._create_instance_with_networks_json([FAKE_NETWORKS[0]])
266
self.assertEquals(response.status_int, 202)
267
self.assertEquals(networks, [FAKE_NETWORKS[0]])
269
def test_create_instance_with_one_network_xml(self):
270
request, response, networks = \
271
self._create_instance_with_networks_xml([FAKE_NETWORKS[0]])
272
self.assertEquals(response.status_int, 202)
273
self.assertEquals(networks, [FAKE_NETWORKS[0]])
275
def test_create_instance_with_two_networks(self):
276
request, response, networks = \
277
self._create_instance_with_networks_json(FAKE_NETWORKS)
278
self.assertEquals(response.status_int, 202)
279
self.assertEquals(networks, FAKE_NETWORKS)
281
def test_create_instance_with_two_networks_xml(self):
282
request, response, networks = \
283
self._create_instance_with_networks_xml(FAKE_NETWORKS)
284
self.assertEquals(response.status_int, 202)
285
self.assertEquals(networks, FAKE_NETWORKS)
287
def test_create_instance_with_duplicate_networks(self):
288
request, response, networks = \
289
self._create_instance_with_networks_json(DUPLICATE_NETWORKS)
290
self.assertEquals(response.status_int, 400)
291
self.assertEquals(networks, None)
293
def test_create_instance_with_duplicate_networks_xml(self):
294
request, response, networks = \
295
self._create_instance_with_networks_xml(DUPLICATE_NETWORKS)
296
self.assertEquals(response.status_int, 400)
297
self.assertEquals(networks, None)
299
def test_create_instance_with_network_no_id(self):
300
body_dict = self._create_networks_request_dict([FAKE_NETWORKS[0]])
301
del body_dict['server']['networks'][0]['uuid']
302
request = self._get_create_request_json(body_dict)
303
compute_api, response = \
304
self._run_create_instance_with_mock_compute_api(request)
305
self.assertEquals(response.status_int, 400)
306
self.assertEquals(compute_api.networks, None)
308
def test_create_instance_with_network_no_id_xml(self):
309
body_dict = self._create_networks_request_dict([FAKE_NETWORKS[0]])
310
request = self._get_create_request_xml(body_dict)
311
uuid = ' uuid="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"'
312
request.body = request.body.replace(uuid, '')
313
compute_api, response = \
314
self._run_create_instance_with_mock_compute_api(request)
315
self.assertEquals(response.status_int, 400)
316
self.assertEquals(compute_api.networks, None)
318
def test_create_instance_with_network_invalid_id(self):
319
request, response, networks = \
320
self._create_instance_with_networks_json(INVALID_NETWORKS)
321
self.assertEquals(response.status_int, 400)
322
self.assertEquals(networks, None)
324
def test_create_instance_with_network_invalid_id_xml(self):
325
request, response, networks = \
326
self._create_instance_with_networks_xml(INVALID_NETWORKS)
327
self.assertEquals(response.status_int, 400)
328
self.assertEquals(networks, None)
330
def test_create_instance_with_network_empty_fixed_ip(self):
331
networks = [('1', '')]
332
request, response, networks = \
333
self._create_instance_with_networks_json(networks)
334
self.assertEquals(response.status_int, 400)
335
self.assertEquals(networks, None)
337
def test_create_instance_with_network_non_string_fixed_ip(self):
338
networks = [('1', 12345)]
339
request, response, networks = \
340
self._create_instance_with_networks_json(networks)
341
self.assertEquals(response.status_int, 400)
342
self.assertEquals(networks, None)
344
def test_create_instance_with_network_empty_fixed_ip_xml(self):
345
networks = [('1', '')]
346
request, response, networks = \
347
self._create_instance_with_networks_xml(networks)
348
self.assertEquals(response.status_int, 400)
349
self.assertEquals(networks, None)
351
def test_create_instance_with_network_no_fixed_ip(self):
352
body_dict = self._create_networks_request_dict([FAKE_NETWORKS[0]])
353
del body_dict['server']['networks'][0]['fixed_ip']
354
request = self._get_create_request_json(body_dict)
355
compute_api, response = \
356
self._run_create_instance_with_mock_compute_api(request)
357
self.assertEquals(response.status_int, 202)
358
self.assertEquals(compute_api.networks,
359
[('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', None)])
361
def test_create_instance_with_network_no_fixed_ip_xml(self):
362
body_dict = self._create_networks_request_dict([FAKE_NETWORKS[0]])
363
request = self._get_create_request_xml(body_dict)
364
request.body = request.body.replace(' fixed_ip="10.0.1.12"', '')
365
compute_api, response = \
366
self._run_create_instance_with_mock_compute_api(request)
367
self.assertEquals(response.status_int, 202)
368
self.assertEquals(compute_api.networks,
369
[('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', None)])
371
def test_create_instance_with_userdata(self):
372
user_data_contents = '#!/bin/bash\necho "Oh no!"\n'
373
user_data_contents = base64.b64encode(user_data_contents)
374
request, response, user_data = \
375
self._create_instance_with_user_data_json(user_data_contents)
376
self.assertEquals(response.status_int, 202)
377
self.assertEquals(user_data, user_data_contents)
379
def test_create_instance_with_userdata_none(self):
380
user_data_contents = None
381
request, response, user_data = \
382
self._create_instance_with_user_data_json(user_data_contents)
383
self.assertEquals(response.status_int, 202)
384
self.assertEquals(user_data, user_data_contents)
386
def test_create_instance_with_userdata_with_non_b64_content(self):
387
user_data_contents = '#!/bin/bash\necho "Oh no!"\n'
388
request, response, user_data = \
389
self._create_instance_with_user_data_json(user_data_contents)
390
self.assertEquals(response.status_int, 400)
391
self.assertEquals(user_data, None)
393
def test_create_instance_with_security_group_json(self):
394
security_groups = ['test', 'test1']
395
self.stubs.Set(nova.db, 'security_group_get_by_name',
396
return_security_group_get_by_name)
397
self.stubs.Set(nova.db, 'instance_add_security_group',
398
return_instance_add_security_group)
399
self._setup_mock_network_api()
400
body_dict = self._create_security_group_request_dict(security_groups)
401
request = self._get_create_request_json(body_dict)
402
compute_api, response = \
403
self._run_create_instance_with_mock_compute_api(request)
404
self.assertEquals(response.status_int, 202)
406
def test_get_server_by_id_verify_security_groups_json(self):
407
self.stubs.Set(nova.db, 'instance_get', return_server_by_id)
408
self._setup_mock_network_api()
409
req = webob.Request.blank('/v2/123/os-create-server-ext/1')
410
req.headers['Content-Type'] = 'application/json'
411
response = req.get_response(fakes.wsgi_app())
412
self.assertEquals(response.status_int, 200)
413
res_dict = json.loads(response.body)
414
expected_security_group = [{"name": "test"}]
415
self.assertEquals(res_dict['server']['security_groups'],
416
expected_security_group)
418
def test_get_server_by_id_verify_security_groups_xml(self):
419
self.stubs.Set(nova.db, 'instance_get', return_server_by_id)
420
self._setup_mock_network_api()
421
req = webob.Request.blank('/v2/123/os-create-server-ext/1')
422
req.headers['Accept'] = 'application/xml'
423
response = req.get_response(fakes.wsgi_app())
424
self.assertEquals(response.status_int, 200)
425
dom = minidom.parseString(response.body)
426
server = dom.childNodes[0]
427
sec_groups = server.getElementsByTagName('security_groups')[0]
428
sec_group = sec_groups.getElementsByTagName('security_group')[0]
429
self.assertEqual(INSTANCE['security_groups'][0]['name'],
430
sec_group.getAttribute("name"))
added
removed
.pc/security-fix-tenant-bypass.patch/nova/tests/api/openstack/v2/contrib/test_createserverext.py
