1
# Copyright 2011 OpenStack LLC.
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
8
# http://www.apache.org/licenses/LICENSE-2.0
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
20
from lxml import etree
22
from nova.api import auth
23
from nova.api.openstack import v2
24
from nova.api.openstack.v2 import wsgi
25
from nova.api.openstack.v2.contrib import cloudpipe
26
from nova.auth import manager
27
from nova.cloudpipe import pipelib
28
from nova import context
29
from nova import crypto
31
from nova import flags
33
from nova.tests.api.openstack import fakes
34
from nova import utils
37
EMPTY_INSTANCE_LIST = True
41
class FakeProject(object):
42
def __init__(self, id, name, manager, desc, members, ip, port):
45
self.project_manager_id = manager
46
self.description = desc
47
self.member_ids = members
52
def fake_vpn_instance():
53
return {'id': 7, 'image_id': FLAGS.vpn_image_id, 'vm_state': 'active',
54
'created_at': utils.parse_strtime('1981-10-20T00:00:00.000000'),
58
def fake_vpn_instance_low_id():
59
return {'id': 4, 'image_id': FLAGS.vpn_image_id, 'vm_state': 'active',
60
'created_at': utils.parse_strtime('1981-10-20T00:00:00.000000')}
64
proj = FakeProject(1, '1', 'fakeuser', '', [1], '127.0.0.1', 22)
68
def db_instance_get_all_by_project(self, project_id):
69
if EMPTY_INSTANCE_LIST:
72
return [fake_vpn_instance()]
75
def db_security_group_exists(context, project_id, group_name):
80
def pipelib_launch_vpn_instance(self, project_id, user_id):
81
global EMPTY_INSTANCE_LIST
82
EMPTY_INSTANCE_LIST = False
85
def auth_manager_get_project(self, project_id):
89
def auth_manager_get_projects(self):
90
return [fake_project()]
93
def utils_vpn_ping(addr, port, timoeout=0.05, session_id=None):
97
def better_not_call_this(*args, **kwargs):
98
raise Exception("You should not have done that")
101
class FakeAuthManager(object):
102
def get_projects(self):
103
return [fake_project()]
105
def get_project(self, project_id):
106
return fake_project()
109
class CloudpipeTest(test.TestCase):
112
super(CloudpipeTest, self).setUp()
113
self.flags(allow_admin_api=True)
114
self.app = fakes.wsgi_app()
115
inner_app = v2.APIRouter()
116
self.context = context.RequestContext('fake', 'fake', is_admin=True)
117
self.app = auth.InjectContext(self.context, inner_app)
118
route = inner_app.map.match('/1234/os-cloudpipe')
119
self.controller = route['controller'].controller
120
fakes.stub_out_networking(self.stubs)
121
fakes.stub_out_rate_limiting(self.stubs)
122
self.stubs.Set(db, "instance_get_all_by_project",
123
db_instance_get_all_by_project)
124
self.stubs.Set(db, "security_group_exists",
125
db_security_group_exists)
126
self.stubs.SmartSet(self.controller.cloudpipe, "launch_vpn_instance",
127
pipelib_launch_vpn_instance)
128
#self.stubs.SmartSet(self.controller.auth_manager, "get_project",
129
# auth_manager_get_project)
130
#self.stubs.SmartSet(self.controller.auth_manager, "get_projects",
131
# auth_manager_get_projects)
132
# NOTE(todd): The above code (just setting the stub, not invoking it)
133
# causes failures in AuthManagerLdapTestCase. So use a fake object.
134
self.controller.auth_manager = FakeAuthManager()
135
self.stubs.Set(utils, 'vpn_ping', utils_vpn_ping)
136
global EMPTY_INSTANCE_LIST
137
EMPTY_INSTANCE_LIST = True
139
def test_cloudpipe_list_none_running(self):
140
"""Should still get an entry per-project, just less descriptive."""
141
req = webob.Request.blank('/fake/os-cloudpipe')
142
res = req.get_response(self.app)
143
self.assertEqual(res.status_int, 200)
144
res_dict = json.loads(res.body)
145
response = {'cloudpipes': [{'project_id': 1, 'public_ip': '127.0.0.1',
146
'public_port': 22, 'state': 'pending'}]}
147
self.assertEqual(res_dict, response)
149
def test_cloudpipe_list(self):
150
global EMPTY_INSTANCE_LIST
151
EMPTY_INSTANCE_LIST = False
152
req = webob.Request.blank('/fake/os-cloudpipe')
153
res = req.get_response(self.app)
154
self.assertEqual(res.status_int, 200)
155
res_dict = json.loads(res.body)
156
response = {'cloudpipes': [{'project_id': 1, 'public_ip': '127.0.0.1',
157
'public_port': 22, 'state': 'running',
159
'created_at': '1981-10-20T00:00:00Z'}]}
160
self.assertEqual(res_dict, response)
162
def test_cloudpipe_create(self):
163
body = {'cloudpipe': {'project_id': 1}}
164
req = webob.Request.blank('/fake/os-cloudpipe')
166
req.body = json.dumps(body)
167
req.headers['Content-Type'] = 'application/json'
168
res = req.get_response(self.app)
169
self.assertEqual(res.status_int, 200)
170
res_dict = json.loads(res.body)
171
response = {'instance_id': 7777}
172
self.assertEqual(res_dict, response)
174
def test_cloudpipe_create_already_running(self):
175
global EMPTY_INSTANCE_LIST
176
EMPTY_INSTANCE_LIST = False
177
self.stubs.SmartSet(self.controller.cloudpipe, 'launch_vpn_instance',
178
better_not_call_this)
179
body = {'cloudpipe': {'project_id': 1}}
180
req = webob.Request.blank('/fake/os-cloudpipe')
182
req.body = json.dumps(body)
183
req.headers['Content-Type'] = 'application/json'
184
res = req.get_response(self.app)
185
self.assertEqual(res.status_int, 200)
186
res_dict = json.loads(res.body)
187
response = {'instance_id': 7777}
188
self.assertEqual(res_dict, response)
191
class CloudpipesXMLSerializerTest(test.TestCase):
192
def test_default_serializer(self):
193
serializer = cloudpipe.CloudpipeTemplate()
194
exemplar = dict(cloudpipe=dict(instance_id='1234-1234-1234-1234'))
195
text = serializer.serialize(exemplar)
196
tree = etree.fromstring(text)
197
self.assertEqual('cloudpipe', tree.tag)
199
self.assertTrue(child.tag in exemplar['cloudpipe'])
200
self.assertEqual(child.text, exemplar['cloudpipe'][child.tag])
202
def test_index_serializer(self):
203
serializer = cloudpipe.CloudpipesTemplate()
204
exemplar = dict(cloudpipes=[
209
instance_id='1234-1234-1234-1234',
210
created_at=utils.isotime(datetime.datetime.utcnow()),
217
text = serializer.serialize(exemplar)
218
tree = etree.fromstring(text)
219
self.assertEqual('cloudpipes', tree.tag)
220
self.assertEqual(len(exemplar['cloudpipes']), len(tree))
221
for idx, cl_pipe in enumerate(tree):
222
self.assertEqual('cloudpipe', cl_pipe.tag)
223
kp_data = exemplar['cloudpipes'][idx]['cloudpipe']
224
for child in cl_pipe:
225
self.assertTrue(child.tag in kp_data)
226
self.assertEqual(child.text, kp_data[child.tag])
228
def test_deserializer(self):
229
deserializer = wsgi.XMLDeserializer()
230
exemplar = dict(cloudpipe=dict(project_id='4321'))
231
intext = ("<?xml version='1.0' encoding='UTF-8'?>\n"
232
'<cloudpipe><project_id>4321</project_id></cloudpipe>')
233
result = deserializer.deserialize(intext)['body']
234
self.assertEqual(result, exemplar)
added
removed
nova/tests/api/openstack/v2/contrib/test_cloudpipe.py
