← Back to branch summary

~cloudbuilders/nova/os-keypair-integration

« back to all changes in this revision

Viewing changes to nova/tests/api/openstack/contrib/test_createserverext.py

  • Committer: Jesse Andrews
  • Date: 2011-08-26 21:57:53 UTC
  • mfrom: (1455.1.45 nova)
  • Revision ID: anotherjesse@gmail.com-20110826215753-0sfp6dubujsl23wa
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/api/openstack/contrib/test_createserverext.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright 2010-2011 OpenStack LLC.
 
4
# All Rights Reserved.
 
5
#
 
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
 
9
#
 
10
#         http://www.apache.org/licenses/LICENSE-2.0
 
11
#
 
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
 
16
#    under the License.
 
17
 
 
18
import base64
 
19
import json
 
20
import unittest
 
21
from xml.dom import minidom
 
22
 
 
23
import stubout
 
24
import webob
 
25
 
 
26
from nova import exception
 
27
from nova import flags
 
28
from nova import test
 
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
 
34
 
 
35
import nova.scheduler.api
 
36
import nova.image.fake
 
37
import nova.rpc
 
38
from nova.tests.api.openstack import fakes
 
39
 
 
40
 
 
41
FLAGS = flags.FLAGS
 
42
FLAGS.verbose = True
 
43
 
 
44
FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
 
45
 
 
46
FAKE_NETWORKS = [('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12'),
 
47
                 ('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', '10.0.2.12')]
 
48
 
 
49
DUPLICATE_NETWORKS = [('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12'),
 
50
                      ('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '10.0.1.12')]
 
51
 
 
52
INVALID_NETWORKS = [('invalid', 'invalid-ip-address')]
 
53
 
 
54
 
 
55
class CreateserverextTest(test.TestCase):
 
56
 
 
57
    def setUp(self):
 
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
 
66
 
 
67
    def tearDown(self):
 
68
        self.stubs.UnsetAll()
 
69
        FLAGS.allow_admin_api = self.allow_admin
 
70
        super(CreateserverextTest, self).tearDown()
 
71
 
 
72
    def _setup_mock_compute_api(self):
 
73
 
 
74
        class MockComputeAPI(nova.compute.API):
 
75
 
 
76
            def __init__(self):
 
77
                self.injected_files = None
 
78
                self.networks = None
 
79
 
 
80
            def create(self, *args, **kwargs):
 
81
                if 'injected_files' in kwargs:
 
82
                    self.injected_files = kwargs['injected_files']
 
83
                else:
 
84
                    self.injected_files = None
 
85
 
 
86
                if 'requested_networks' in kwargs:
 
87
                    self.networks = kwargs['requested_networks']
 
88
                else:
 
89
                    self.networks = None
 
90
                return [{'id': '1234', 'display_name': 'fakeinstance',
 
91
                         'uuid': FAKE_UUID,
 
92
                         'created_at': "",
 
93
                         'updated_at': ""}]
 
94
 
 
95
            def set_admin_password(self, *args, **kwargs):
 
96
                pass
 
97
 
 
98
        def make_stub_method(canned_return):
 
99
            def stub_method(*args, **kwargs):
 
100
                return canned_return
 
101
            return stub_method
 
102
 
 
103
        compute_api = MockComputeAPI()
 
104
        self.stubs.Set(nova.compute, 'API', make_stub_method(compute_api))
 
105
        self.stubs.Set(
 
106
            nova.api.openstack.create_instance_helper.CreateInstanceHelper,
 
107
            '_get_kernel_ramdisk_from_image', make_stub_method((1, 1)))
 
108
        return compute_api
 
109
 
 
110
    def _create_networks_request_dict(self, networks):
 
111
        server = {}
 
112
        server['name'] = 'new-server-test'
 
113
        server['imageRef'] = 1
 
114
        server['flavorRef'] = 1
 
115
        if networks is not None:
 
116
            network_list = []
 
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}
 
121
 
 
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'
 
125
        req.method = 'POST'
 
126
        req.body = json.dumps(body_dict)
 
127
        return req
 
128
 
 
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
 
133
 
 
134
    def _format_xml_request_body(self, body_dict):
 
135
        server = body_dict['server']
 
136
        body_parts = []
 
137
        body_parts.extend([
 
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>'
 
161
                                  % item)
 
162
            body_parts.append('</networks>')
 
163
        body_parts.append('</server>')
 
164
        return ''.join(body_parts)
 
165
 
 
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'
 
170
        req.method = 'POST'
 
171
        req.body = self._format_xml_request_body(body_dict)
 
172
        return req
 
173
 
 
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
 
180
 
 
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
 
187
 
 
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)
 
193
 
 
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)
 
199
 
 
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]])
 
205
 
 
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]])
 
211
 
 
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)
 
217
 
 
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)
 
223
 
 
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)
 
229
 
 
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)
 
235
 
 
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)
 
244
 
 
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)
 
254
 
 
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)
 
260
 
 
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)
 
266
 
 
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)
 
273
 
 
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)
 
280
 
 
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)
 
287
 
 
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)])
 
297
 
 
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)])