1
# vim: tabstop=4 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
26
from nova import exception
27
from nova import flags
29
from nova import utils
30
import nova.api.openstack
31
from nova.api.openstack import servers
32
from nova.api.openstack.contrib import createserverext
33
import nova.compute.api
35
import nova.scheduler.api
36
import nova.image.fake
38
from nova.tests.api.openstack import fakes
44
FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
46
FAKE_NETWORKS = [('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12'),
47
('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', '10.0.2.12')]
49
DUPLICATE_NETWORKS = [('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12'),
50
('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12')]
52
INVALID_NETWORKS = [('invalid', 'invalid-ip-address')]
55
class CreateserverextTest(test.TestCase):
58
super(CreateserverextTest, self).setUp()
59
self.stubs = stubout.StubOutForTesting()
60
fakes.FakeAuthManager.auth_data = {}
61
fakes.FakeAuthDatabase.data = {}
62
fakes.stub_out_auth(self.stubs)
63
fakes.stub_out_image_service(self.stubs)
64
fakes.stub_out_key_pair_funcs(self.stubs)
65
self.allow_admin = FLAGS.allow_admin_api
69
FLAGS.allow_admin_api = self.allow_admin
70
super(CreateserverextTest, self).tearDown()
72
def _setup_mock_compute_api(self):
74
class MockComputeAPI(nova.compute.API):
77
self.injected_files = None
80
def create(self, *args, **kwargs):
81
if 'injected_files' in kwargs:
82
self.injected_files = kwargs['injected_files']
84
self.injected_files = None
86
if 'requested_networks' in kwargs:
87
self.networks = kwargs['requested_networks']
90
return [{'id': '1234', 'display_name': 'fakeinstance',
95
def set_admin_password(self, *args, **kwargs):
98
def make_stub_method(canned_return):
99
def stub_method(*args, **kwargs):
103
compute_api = MockComputeAPI()
104
self.stubs.Set(nova.compute, 'API', make_stub_method(compute_api))
106
nova.api.openstack.create_instance_helper.CreateInstanceHelper,
107
'_get_kernel_ramdisk_from_image', make_stub_method((1, 1)))
110
def _create_networks_request_dict(self, networks):
112
server['name'] = 'new-server-test'
113
server['imageRef'] = 1
114
server['flavorRef'] = 1
115
if networks is not None:
117
for uuid, fixed_ip in networks:
118
network_list.append({'uuid': uuid, 'fixed_ip': fixed_ip})
119
server['networks'] = network_list
120
return {'server': server}
122
def _get_create_request_json(self, body_dict):
123
req = webob.Request.blank('/v1.1/123/os-create-server-ext')
124
req.headers['Content-Type'] = 'application/json'
126
req.body = json.dumps(body_dict)
129
def _run_create_instance_with_mock_compute_api(self, request):
130
compute_api = self._setup_mock_compute_api()
131
response = request.get_response(fakes.wsgi_app())
132
return compute_api, response
134
def _format_xml_request_body(self, body_dict):
135
server = body_dict['server']
138
'<?xml version="1.0" encoding="UTF-8"?>',
139
'<server xmlns="http://docs.rackspacecloud.com/servers/api/v1.1"',
140
' name="%s" imageRef="%s" flavorRef="%s">' % (
141
server['name'], server['imageRef'], server['flavorRef'])])
142
if 'metadata' in server:
143
metadata = server['metadata']
144
body_parts.append('<metadata>')
145
for item in metadata.iteritems():
146
body_parts.append('<meta key="%s">%s</meta>' % item)
147
body_parts.append('</metadata>')
148
if 'personality' in server:
149
personalities = server['personality']
150
body_parts.append('<personality>')
151
for file in personalities:
152
item = (file['path'], file['contents'])
153
body_parts.append('<file path="%s">%s</file>' % item)
154
body_parts.append('</personality>')
155
if 'networks' in server:
156
networks = server['networks']
157
body_parts.append('<networks>')
158
for network in networks:
159
item = (network['uuid'], network['fixed_ip'])
160
body_parts.append('<network uuid="%s" fixed_ip="%s"></network>'
162
body_parts.append('</networks>')
163
body_parts.append('</server>')
164
return ''.join(body_parts)
166
def _get_create_request_xml(self, body_dict):
167
req = webob.Request.blank('/v1.1/123/os-create-server-ext')
168
req.content_type = 'application/xml'
169
req.accept = 'application/xml'
171
req.body = self._format_xml_request_body(body_dict)
174
def _create_instance_with_networks_json(self, networks):
175
body_dict = self._create_networks_request_dict(networks)
176
request = self._get_create_request_json(body_dict)
177
compute_api, response = \
178
self._run_create_instance_with_mock_compute_api(request)
179
return request, response, compute_api.networks
181
def _create_instance_with_networks_xml(self, networks):
182
body_dict = self._create_networks_request_dict(networks)
183
request = self._get_create_request_xml(body_dict)
184
compute_api, response = \
185
self._run_create_instance_with_mock_compute_api(request)
186
return request, response, compute_api.networks
188
def test_create_instance_with_no_networks(self):
189
request, response, networks = \
190
self._create_instance_with_networks_json(networks=None)
191
self.assertEquals(response.status_int, 202)
192
self.assertEquals(networks, None)
194
def test_create_instance_with_no_networks_xml(self):
195
request, response, networks = \
196
self._create_instance_with_networks_xml(networks=None)
197
self.assertEquals(response.status_int, 202)
198
self.assertEquals(networks, None)
200
def test_create_instance_with_one_network(self):
201
request, response, networks = \
202
self._create_instance_with_networks_json([FAKE_NETWORKS[0]])
203
self.assertEquals(response.status_int, 202)
204
self.assertEquals(networks, [FAKE_NETWORKS[0]])
206
def test_create_instance_with_one_network_xml(self):
207
request, response, networks = \
208
self._create_instance_with_networks_xml([FAKE_NETWORKS[0]])
209
self.assertEquals(response.status_int, 202)
210
self.assertEquals(networks, [FAKE_NETWORKS[0]])
212
def test_create_instance_with_two_networks(self):
213
request, response, networks = \
214
self._create_instance_with_networks_json(FAKE_NETWORKS)
215
self.assertEquals(response.status_int, 202)
216
self.assertEquals(networks, FAKE_NETWORKS)
218
def test_create_instance_with_two_networks_xml(self):
219
request, response, networks = \
220
self._create_instance_with_networks_xml(FAKE_NETWORKS)
221
self.assertEquals(response.status_int, 202)
222
self.assertEquals(networks, FAKE_NETWORKS)
224
def test_create_instance_with_duplicate_networks(self):
225
request, response, networks = \
226
self._create_instance_with_networks_json(DUPLICATE_NETWORKS)
227
self.assertEquals(response.status_int, 400)
228
self.assertEquals(networks, None)
230
def test_create_instance_with_duplicate_networks_xml(self):
231
request, response, networks = \
232
self._create_instance_with_networks_xml(DUPLICATE_NETWORKS)
233
self.assertEquals(response.status_int, 400)
234
self.assertEquals(networks, None)
236
def test_create_instance_with_network_no_id(self):
237
body_dict = self._create_networks_request_dict([FAKE_NETWORKS[0]])
238
del body_dict['server']['networks'][0]['uuid']
239
request = self._get_create_request_json(body_dict)
240
compute_api, response = \
241
self._run_create_instance_with_mock_compute_api(request)
242
self.assertEquals(response.status_int, 400)
243
self.assertEquals(compute_api.networks, None)
245
def test_create_instance_with_network_no_id_xml(self):
246
body_dict = self._create_networks_request_dict([FAKE_NETWORKS[0]])
247
request = self._get_create_request_xml(body_dict)
248
uuid = ' uuid="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"'
249
request.body = request.body.replace(uuid, '')
250
compute_api, response = \
251
self._run_create_instance_with_mock_compute_api(request)
252
self.assertEquals(response.status_int, 400)
253
self.assertEquals(compute_api.networks, None)
255
def test_create_instance_with_network_invalid_id(self):
256
request, response, networks = \
257
self._create_instance_with_networks_json(INVALID_NETWORKS)
258
self.assertEquals(response.status_int, 400)
259
self.assertEquals(networks, None)
261
def test_create_instance_with_network_invalid_id_xml(self):
262
request, response, networks = \
263
self._create_instance_with_networks_xml(INVALID_NETWORKS)
264
self.assertEquals(response.status_int, 400)
265
self.assertEquals(networks, None)
267
def test_create_instance_with_network_empty_fixed_ip(self):
268
networks = [('1', '')]
269
request, response, networks = \
270
self._create_instance_with_networks_json(networks)
271
self.assertEquals(response.status_int, 400)
272
self.assertEquals(networks, None)
274
def test_create_instance_with_network_non_string_fixed_ip(self):
275
networks = [('1', 12345)]
276
request, response, networks = \
277
self._create_instance_with_networks_json(networks)
278
self.assertEquals(response.status_int, 400)
279
self.assertEquals(networks, None)
281
def test_create_instance_with_network_empty_fixed_ip_xml(self):
282
networks = [('1', '')]
283
request, response, networks = \
284
self._create_instance_with_networks_xml(networks)
285
self.assertEquals(response.status_int, 400)
286
self.assertEquals(networks, None)
288
def test_create_instance_with_network_no_fixed_ip(self):
289
body_dict = self._create_networks_request_dict([FAKE_NETWORKS[0]])
290
del body_dict['server']['networks'][0]['fixed_ip']
291
request = self._get_create_request_json(body_dict)
292
compute_api, response = \
293
self._run_create_instance_with_mock_compute_api(request)
294
self.assertEquals(response.status_int, 202)
295
self.assertEquals(compute_api.networks,
296
[('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', None)])
298
def test_create_instance_with_network_no_fixed_ip_xml(self):
299
body_dict = self._create_networks_request_dict([FAKE_NETWORKS[0]])
300
request = self._get_create_request_xml(body_dict)
301
request.body = request.body.replace(' fixed_ip="10.0.1.12"', '')
302
compute_api, response = \
303
self._run_create_instance_with_mock_compute_api(request)
304
self.assertEquals(response.status_int, 202)
305
self.assertEquals(compute_api.networks,
306
[('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', None)])