1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2012 United States Government as represented by the
4
# Administrator of the National Aeronautics and Space Administration.
7
# Copyright 2012 Nebula, Inc.
9
# Licensed under the Apache License, Version 2.0 (the "License"); you may
10
# not use this file except in compliance with the License. You may obtain
11
# a copy of the License at
13
# http://www.apache.org/licenses/LICENSE-2.0
15
# Unless required by applicable law or agreed to in writing, software
16
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18
# License for the specific language governing permissions and limitations
21
from django import http
22
from django.conf import settings
23
from django.core.urlresolvers import reverse
27
from horizon import api
28
from horizon import test
29
from .tables import SecurityGroupsTable, RulesTable
31
INDEX_URL = reverse('horizon:nova:access_and_security:index')
33
reverse('horizon:nova:access_and_security:security_groups:create')
36
def strip_absolute_base(uri):
37
return uri.split(settings.TESTSERVER, 1)[-1]
40
class SecurityGroupsViewTests(test.TestCase):
42
super(SecurityGroupsViewTests, self).setUp()
43
sec_group = self.security_groups.first()
44
self.edit_url = reverse('horizon:nova:access_and_security:'
45
'security_groups:edit_rules',
48
def test_create_security_groups_get(self):
49
res = self.client.get(SG_CREATE_URL)
50
self.assertTemplateUsed(res,
51
'nova/access_and_security/security_groups/create.html')
53
def test_create_security_groups_post(self):
54
sec_group = self.security_groups.first()
55
self.mox.StubOutWithMock(api, 'security_group_create')
56
api.security_group_create(IsA(http.HttpRequest),
58
sec_group.description).AndReturn(sec_group)
61
formData = {'method': 'CreateGroup',
62
'name': sec_group.name,
63
'description': sec_group.description}
64
res = self.client.post(SG_CREATE_URL, formData)
65
self.assertRedirectsNoFollow(res, INDEX_URL)
67
def test_create_security_groups_post_exception(self):
68
sec_group = self.security_groups.first()
69
self.mox.StubOutWithMock(api, 'security_group_create')
70
api.security_group_create(IsA(http.HttpRequest),
72
sec_group.description) \
73
.AndRaise(self.exceptions.nova)
76
formData = {'method': 'CreateGroup',
77
'name': sec_group.name,
78
'description': sec_group.description}
79
res = self.client.post(SG_CREATE_URL, formData)
80
self.assertMessageCount(error=1)
81
self.assertRedirectsNoFollow(res, INDEX_URL)
83
def test_edit_rules_get(self):
84
sec_group = self.security_groups.first()
85
sec_group_list = self.security_groups.list()
87
self.mox.StubOutWithMock(api, 'security_group_get')
88
api.security_group_get(IsA(http.HttpRequest),
89
sec_group.id).AndReturn(sec_group)
90
self.mox.StubOutWithMock(api, 'security_group_list')
91
api.security_group_list(
92
IsA(http.HttpRequest)).AndReturn(sec_group_list)
95
res = self.client.get(self.edit_url)
96
self.assertTemplateUsed(res,
97
'nova/access_and_security/security_groups/edit_rules.html')
98
self.assertItemsEqual(res.context['security_group'].name,
101
def test_edit_rules_get_exception(self):
102
sec_group = self.security_groups.first()
104
self.mox.StubOutWithMock(api, 'security_group_get')
105
self.mox.StubOutWithMock(api, 'security_group_list')
107
api.security_group_get(IsA(http.HttpRequest),
108
sec_group.id).AndRaise(self.exceptions.nova)
111
res = self.client.get(self.edit_url)
112
self.assertRedirectsNoFollow(res, INDEX_URL)
114
def test_edit_rules_add_rule_cidr(self):
115
sec_group = self.security_groups.first()
116
sec_group_list = self.security_groups.list()
117
rule = self.security_group_rules.first()
119
self.mox.StubOutWithMock(api, 'security_group_rule_create')
120
self.mox.StubOutWithMock(api, 'security_group_list')
121
api.security_group_rule_create(IsA(http.HttpRequest),
126
rule.ip_range['cidr'],
127
None).AndReturn(rule)
128
api.security_group_list(
129
IsA(http.HttpRequest)).AndReturn(sec_group_list)
132
formData = {'method': 'AddRule',
133
'security_group_id': sec_group.id,
134
'from_port': rule.from_port,
135
'to_port': rule.to_port,
136
'ip_protocol': rule.ip_protocol,
137
'cidr': rule.ip_range['cidr'],
139
res = self.client.post(self.edit_url, formData)
140
self.assertRedirectsNoFollow(res, INDEX_URL)
142
def test_edit_rules_add_rule_cidr_and_source_group(self):
143
sec_group = self.security_groups.first()
144
sec_group_other = self.security_groups.get(id=2)
145
sec_group_list = self.security_groups.list()
146
rule = self.security_group_rules.first()
148
self.mox.StubOutWithMock(api, 'security_group_get')
149
self.mox.StubOutWithMock(api, 'security_group_list')
150
api.security_group_get(IsA(http.HttpRequest),
151
sec_group.id).AndReturn(sec_group)
152
api.security_group_list(
153
IsA(http.HttpRequest)).AndReturn(sec_group_list)
156
formData = {'method': 'AddRule',
157
'security_group_id': sec_group.id,
158
'from_port': rule.from_port,
159
'to_port': rule.to_port,
160
'ip_protocol': rule.ip_protocol,
161
'cidr': "127.0.0.1/32",
162
'source_group': sec_group_other.id}
163
res = self.client.post(self.edit_url, formData)
164
self.assertNoMessages()
165
msg = 'Either CIDR or Source Group may be specified, but not both.'
166
self.assertFormErrors(res, count=1, message=msg)
168
def test_edit_rules_add_rule_self_as_source_group(self):
169
sec_group = self.security_groups.first()
170
sec_group_list = self.security_groups.list()
171
rule = self.security_group_rules.get(id=3)
173
self.mox.StubOutWithMock(api, 'security_group_rule_create')
174
self.mox.StubOutWithMock(api, 'security_group_list')
175
api.security_group_rule_create(IsA(http.HttpRequest),
181
u'%s' % sec_group.id).AndReturn(rule)
182
api.security_group_list(
183
IsA(http.HttpRequest)).AndReturn(sec_group_list)
186
formData = {'method': 'AddRule',
187
'security_group_id': sec_group.id,
188
'from_port': rule.from_port,
189
'to_port': rule.to_port,
190
'ip_protocol': rule.ip_protocol,
192
'source_group': sec_group.id}
193
res = self.client.post(self.edit_url, formData)
194
self.assertRedirectsNoFollow(res, INDEX_URL)
196
def test_edit_rules_invalid_port_range(self):
197
sec_group = self.security_groups.first()
198
sec_group_list = self.security_groups.list()
199
rule = self.security_group_rules.first()
201
self.mox.StubOutWithMock(api, 'security_group_get')
202
api.security_group_get(IsA(http.HttpRequest),
203
sec_group.id).AndReturn(sec_group)
204
self.mox.StubOutWithMock(api, 'security_group_list')
205
api.security_group_list(
206
IsA(http.HttpRequest)).AndReturn(sec_group_list)
209
formData = {'method': 'AddRule',
210
'security_group_id': sec_group.id,
211
'from_port': rule.from_port,
212
'to_port': int(rule.from_port) - 1,
213
'ip_protocol': rule.ip_protocol,
214
'cidr': rule.ip_range['cidr'],
216
res = self.client.post(self.edit_url, formData)
217
self.assertNoMessages()
218
self.assertContains(res, "greater than or equal to")
220
@test.create_stubs({api: ('security_group_get', 'security_group_list')})
221
def test_edit_rules_invalid_icmp_rule(self):
222
sec_group = self.security_groups.first()
223
sec_group_list = self.security_groups.list()
224
icmp_rule = self.security_group_rules.list()[1]
226
api.security_group_get(IsA(http.HttpRequest),
227
sec_group.id).AndReturn(sec_group)
228
api.security_group_list(
229
IsA(http.HttpRequest)).AndReturn(sec_group_list)
230
api.security_group_get(IsA(http.HttpRequest),
231
sec_group.id).AndReturn(sec_group)
232
api.security_group_list(
233
IsA(http.HttpRequest)).AndReturn(sec_group_list)
234
api.security_group_get(IsA(http.HttpRequest),
235
sec_group.id).AndReturn(sec_group)
236
api.security_group_list(
237
IsA(http.HttpRequest)).AndReturn(sec_group_list)
238
api.security_group_get(IsA(http.HttpRequest),
239
sec_group.id).AndReturn(sec_group)
240
api.security_group_list(
241
IsA(http.HttpRequest)).AndReturn(sec_group_list)
244
formData = {'method': 'AddRule',
245
'security_group_id': sec_group.id,
247
'to_port': icmp_rule.to_port,
248
'ip_protocol': icmp_rule.ip_protocol,
249
'cidr': icmp_rule.ip_range['cidr'],
251
res = self.client.post(self.edit_url, formData)
252
self.assertNoMessages()
253
self.assertContains(res, "The ICMP type not in range (-1, 255)")
255
formData = {'method': 'AddRule',
256
'security_group_id': sec_group.id,
257
'from_port': icmp_rule.from_port,
259
'ip_protocol': icmp_rule.ip_protocol,
260
'cidr': icmp_rule.ip_range['cidr'],
262
res = self.client.post(self.edit_url, formData)
263
self.assertNoMessages()
264
self.assertContains(res, "The ICMP code not in range (-1, 255)")
266
formData = {'method': 'AddRule',
267
'security_group_id': sec_group.id,
268
'from_port': icmp_rule.from_port,
270
'ip_protocol': icmp_rule.ip_protocol,
271
'cidr': icmp_rule.ip_range['cidr'],
273
res = self.client.post(self.edit_url, formData)
274
self.assertNoMessages()
275
self.assertContains(res, "The ICMP code is invalid")
277
formData = {'method': 'AddRule',
278
'security_group_id': sec_group.id,
280
'to_port': icmp_rule.to_port,
281
'ip_protocol': icmp_rule.ip_protocol,
282
'cidr': icmp_rule.ip_range['cidr'],
284
res = self.client.post(self.edit_url, formData)
285
self.assertNoMessages()
286
self.assertContains(res, "The ICMP type is invalid")
288
def test_edit_rules_add_rule_exception(self):
289
sec_group = self.security_groups.first()
290
sec_group_list = self.security_groups.list()
291
rule = self.security_group_rules.first()
293
self.mox.StubOutWithMock(api, 'security_group_rule_create')
294
self.mox.StubOutWithMock(api, 'security_group_list')
295
api.security_group_rule_create(IsA(http.HttpRequest),
300
rule.ip_range['cidr'],
301
None).AndRaise(self.exceptions.nova)
302
api.security_group_list(
303
IsA(http.HttpRequest)).AndReturn(sec_group_list)
306
formData = {'method': 'AddRule',
307
'security_group_id': sec_group.id,
308
'from_port': rule.from_port,
309
'to_port': rule.to_port,
310
'ip_protocol': rule.ip_protocol,
311
'cidr': rule.ip_range['cidr'],
313
res = self.client.post(self.edit_url, formData)
314
self.assertRedirectsNoFollow(res, INDEX_URL)
316
def test_edit_rules_delete_rule(self):
317
sec_group = self.security_groups.first()
318
rule = self.security_group_rules.first()
320
self.mox.StubOutWithMock(api, 'security_group_rule_delete')
321
api.security_group_rule_delete(IsA(http.HttpRequest), rule.id)
324
form_data = {"action": "rules__delete__%s" % rule.id}
325
req = self.factory.post(self.edit_url, form_data)
326
table = RulesTable(req, sec_group.rules)
327
handled = table.maybe_handle()
328
self.assertEqual(strip_absolute_base(handled['location']), INDEX_URL)
330
def test_edit_rules_delete_rule_exception(self):
331
rule = self.security_group_rules.first()
333
self.mox.StubOutWithMock(api, 'security_group_rule_delete')
334
api.security_group_rule_delete(IsA(http.HttpRequest),
335
rule.id).AndRaise(self.exceptions.nova)
338
form_data = {"action": "rules__delete__%s" % rule.id}
339
req = self.factory.post(self.edit_url, form_data)
340
table = RulesTable(req, self.security_group_rules.list())
341
handled = table.maybe_handle()
342
self.assertEqual(strip_absolute_base(handled['location']),
345
def test_delete_group(self):
346
sec_group = self.security_groups.get(name="other_group")
348
self.mox.StubOutWithMock(api, 'security_group_delete')
349
api.security_group_delete(IsA(http.HttpRequest), sec_group.id)
352
form_data = {"action": "security_groups__delete__%s" % sec_group.id}
353
req = self.factory.post(INDEX_URL, form_data)
354
table = SecurityGroupsTable(req, self.security_groups.list())
355
handled = table.maybe_handle()
356
self.assertEqual(strip_absolute_base(handled['location']),
359
def test_delete_group_exception(self):
360
sec_group = self.security_groups.get(name="other_group")
362
self.mox.StubOutWithMock(api, 'security_group_delete')
363
api.security_group_delete(IsA(http.HttpRequest),
364
sec_group.id).AndRaise(self.exceptions.nova)
368
form_data = {"action": "security_groups__delete__%s" % sec_group.id}
369
req = self.factory.post(INDEX_URL, form_data)
370
table = SecurityGroupsTable(req, self.security_groups.list())
371
handled = table.maybe_handle()
373
self.assertEqual(strip_absolute_base(handled['location']),