Skip to content

Commit b4421e3

Browse files
Jenkinsopenstack-gerrit
Jenkins
authored and
openstack-gerrit
committed
Merge "Properly handle port arguments for ICMP"
2 parents 24ca952 + e6706f2 commit b4421e3

2 files changed

Lines changed: 345 additions & 4 deletions

File tree

‎openstackclient/compute/v2/security_group.py‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -50,10 +50,10 @@ def _xform_security_group_rule(sgroup):
5050
info['ip_range'] = info['ip_range']['cidr']
5151
else:
5252
info['ip_range'] = ''
53-
if info['ip_protocol'] == 'icmp':
54-
info['port_range'] = ''
55-
elif info['ip_protocol'] is None:
53+
if info['ip_protocol'] is None:
5654
info['ip_protocol'] = ''
55+
elif info['ip_protocol'].lower() == 'icmp':
56+
info['port_range'] = ''
5757
return info
5858

5959

@@ -307,7 +307,10 @@ def take_action(self, parsed_args):
307307
compute_client.security_groups,
308308
parsed_args.group,
309309
)
310-
from_port, to_port = parsed_args.dst_port
310+
if parsed_args.proto.lower() == 'icmp':
311+
from_port, to_port = -1, -1
312+
else:
313+
from_port, to_port = parsed_args.dst_port
311314
data = compute_client.security_group_rules.create(
312315
group.id,
313316
parsed_args.proto,

‎openstackclient/tests/compute/v2/test_security_group_rule.py‎

Lines changed: 338 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,338 @@
1+
# Licensed under the Apache License, Version 2.0 (the "License"); you may
2+
# not use this file except in compliance with the License. You may obtain
3+
# a copy of the License at
4+
#
5+
# http://www.apache.org/licenses/LICENSE-2.0
6+
#
7+
# Unless required by applicable law or agreed to in writing, software
8+
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9+
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10+
# License for the specific language governing permissions and limitations
11+
# under the License.
12+
#
13+
14+
import copy
15+
import mock
16+
17+
from openstackclient.compute.v2 import security_group
18+
from openstackclient.tests.compute.v2 import fakes as compute_fakes
19+
from openstackclient.tests import fakes
20+
from openstackclient.tests.identity.v2_0 import fakes as identity_fakes
21+
22+
23+
security_group_id = '11'
24+
security_group_name = 'wide-open'
25+
security_group_description = 'nothing but net'
26+
27+
security_group_rule_id = '1'
28+
29+
SECURITY_GROUP = {
30+
'id': security_group_id,
31+
'name': security_group_name,
32+
'description': security_group_description,
33+
'tenant_id': identity_fakes.project_id,
34+
}
35+
36+
SECURITY_GROUP_RULE = {
37+
'id': security_group_rule_id,
38+
'group': {},
39+
'ip_protocol': 'tcp',
40+
'ip_range': '0.0.0.0/0',
41+
'parent_group_id': security_group_id,
42+
'from_port': 0,
43+
'to_port': 0,
44+
}
45+
46+
SECURITY_GROUP_RULE_ICMP = {
47+
'id': security_group_rule_id,
48+
'group': {},
49+
'ip_protocol': 'icmp',
50+
'ip_range': '0.0.0.0/0',
51+
'parent_group_id': security_group_id,
52+
'from_port': -1,
53+
'to_port': -1,
54+
}
55+
56+
57+
class FakeSecurityGroupRuleResource(fakes.FakeResource):
58+
59+
def get_keys(self):
60+
return {'property': 'value'}
61+
62+
63+
class TestSecurityGroupRule(compute_fakes.TestComputev2):
64+
65+
def setUp(self):
66+
super(TestSecurityGroupRule, self).setUp()
67+
68+
self.secgroups_mock = mock.Mock()
69+
self.secgroups_mock.resource_class = fakes.FakeResource(None, {})
70+
self.app.client_manager.compute.security_groups = self.secgroups_mock
71+
self.secgroups_mock.reset_mock()
72+
73+
self.sg_rules_mock = mock.Mock()
74+
self.sg_rules_mock.resource_class = fakes.FakeResource(None, {})
75+
self.app.client_manager.compute.security_group_rules = \
76+
self.sg_rules_mock
77+
self.sg_rules_mock.reset_mock()
78+
79+
80+
class TestSecurityGroupRuleCreate(TestSecurityGroupRule):
81+
82+
def setUp(self):
83+
super(TestSecurityGroupRuleCreate, self).setUp()
84+
85+
self.secgroups_mock.get.return_value = FakeSecurityGroupRuleResource(
86+
None,
87+
copy.deepcopy(SECURITY_GROUP),
88+
loaded=True,
89+
)
90+
91+
# Get the command object to test
92+
self.cmd = security_group.CreateSecurityGroupRule(self.app, None)
93+
94+
def test_security_group_rule_create_no_options(self):
95+
self.sg_rules_mock.create.return_value = FakeSecurityGroupRuleResource(
96+
None,
97+
copy.deepcopy(SECURITY_GROUP_RULE),
98+
loaded=True,
99+
)
100+
101+
arglist = [
102+
security_group_name,
103+
]
104+
verifylist = [
105+
('group', security_group_name),
106+
]
107+
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
108+
109+
# DisplayCommandBase.take_action() returns two tuples
110+
columns, data = self.cmd.take_action(parsed_args)
111+
112+
# SecurityGroupManager.create(name, description)
113+
self.sg_rules_mock.create.assert_called_with(
114+
security_group_id,
115+
'tcp',
116+
0,
117+
0,
118+
'0.0.0.0/0',
119+
)
120+
121+
collist = (
122+
'group',
123+
'id',
124+
'ip_protocol',
125+
'ip_range',
126+
'parent_group_id',
127+
'port_range',
128+
)
129+
self.assertEqual(collist, columns)
130+
datalist = (
131+
{},
132+
security_group_rule_id,
133+
'tcp',
134+
'',
135+
security_group_id,
136+
'0:0',
137+
)
138+
self.assertEqual(datalist, data)
139+
140+
def test_security_group_rule_create_ftp(self):
141+
sg_rule = copy.deepcopy(SECURITY_GROUP_RULE)
142+
sg_rule['from_port'] = 20
143+
sg_rule['to_port'] = 21
144+
self.sg_rules_mock.create.return_value = FakeSecurityGroupRuleResource(
145+
None,
146+
sg_rule,
147+
loaded=True,
148+
)
149+
150+
arglist = [
151+
security_group_name,
152+
'--dst-port', '20:21',
153+
]
154+
verifylist = [
155+
('group', security_group_name),
156+
('dst_port', (20, 21)),
157+
]
158+
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
159+
160+
# DisplayCommandBase.take_action() returns two tuples
161+
columns, data = self.cmd.take_action(parsed_args)
162+
163+
# SecurityGroupManager.create(name, description)
164+
self.sg_rules_mock.create.assert_called_with(
165+
security_group_id,
166+
'tcp',
167+
20,
168+
21,
169+
'0.0.0.0/0',
170+
)
171+
172+
collist = (
173+
'group',
174+
'id',
175+
'ip_protocol',
176+
'ip_range',
177+
'parent_group_id',
178+
'port_range',
179+
)
180+
self.assertEqual(collist, columns)
181+
datalist = (
182+
{},
183+
security_group_rule_id,
184+
'tcp',
185+
'',
186+
security_group_id,
187+
'20:21',
188+
)
189+
self.assertEqual(datalist, data)
190+
191+
def test_security_group_rule_create_ssh(self):
192+
sg_rule = copy.deepcopy(SECURITY_GROUP_RULE)
193+
sg_rule['from_port'] = 22
194+
sg_rule['to_port'] = 22
195+
self.sg_rules_mock.create.return_value = FakeSecurityGroupRuleResource(
196+
None,
197+
sg_rule,
198+
loaded=True,
199+
)
200+
201+
arglist = [
202+
security_group_name,
203+
'--dst-port', '22',
204+
]
205+
verifylist = [
206+
('group', security_group_name),
207+
('dst_port', (22, 22)),
208+
]
209+
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
210+
211+
# DisplayCommandBase.take_action() returns two tuples
212+
columns, data = self.cmd.take_action(parsed_args)
213+
214+
# SecurityGroupManager.create(name, description)
215+
self.sg_rules_mock.create.assert_called_with(
216+
security_group_id,
217+
'tcp',
218+
22,
219+
22,
220+
'0.0.0.0/0',
221+
)
222+
223+
collist = (
224+
'group',
225+
'id',
226+
'ip_protocol',
227+
'ip_range',
228+
'parent_group_id',
229+
'port_range',
230+
)
231+
self.assertEqual(collist, columns)
232+
datalist = (
233+
{},
234+
security_group_rule_id,
235+
'tcp',
236+
'',
237+
security_group_id,
238+
'22:22',
239+
)
240+
self.assertEqual(datalist, data)
241+
242+
def test_security_group_rule_create_udp(self):
243+
sg_rule = copy.deepcopy(SECURITY_GROUP_RULE)
244+
sg_rule['ip_protocol'] = 'udp'
245+
self.sg_rules_mock.create.return_value = FakeSecurityGroupRuleResource(
246+
None,
247+
sg_rule,
248+
loaded=True,
249+
)
250+
251+
arglist = [
252+
security_group_name,
253+
'--proto', 'udp',
254+
]
255+
verifylist = [
256+
('group', security_group_name),
257+
('proto', 'udp'),
258+
]
259+
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
260+
261+
# DisplayCommandBase.take_action() returns two tuples
262+
columns, data = self.cmd.take_action(parsed_args)
263+
264+
# SecurityGroupManager.create(name, description)
265+
self.sg_rules_mock.create.assert_called_with(
266+
security_group_id,
267+
'udp',
268+
0,
269+
0,
270+
'0.0.0.0/0',
271+
)
272+
273+
collist = (
274+
'group',
275+
'id',
276+
'ip_protocol',
277+
'ip_range',
278+
'parent_group_id',
279+
'port_range',
280+
)
281+
self.assertEqual(collist, columns)
282+
datalist = (
283+
{},
284+
security_group_rule_id,
285+
'udp',
286+
'',
287+
security_group_id,
288+
'0:0',
289+
)
290+
self.assertEqual(datalist, data)
291+
292+
def test_security_group_rule_create_icmp(self):
293+
self.sg_rules_mock.create.return_value = FakeSecurityGroupRuleResource(
294+
None,
295+
copy.deepcopy(SECURITY_GROUP_RULE_ICMP),
296+
loaded=True,
297+
)
298+
299+
arglist = [
300+
security_group_name,
301+
'--proto', 'ICMP',
302+
]
303+
verifylist = [
304+
('group', security_group_name),
305+
('proto', 'ICMP'),
306+
]
307+
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
308+
309+
# DisplayCommandBase.take_action() returns two tuples
310+
columns, data = self.cmd.take_action(parsed_args)
311+
312+
# SecurityGroupManager.create(name, description)
313+
self.sg_rules_mock.create.assert_called_with(
314+
security_group_id,
315+
'ICMP',
316+
-1,
317+
-1,
318+
'0.0.0.0/0',
319+
)
320+
321+
collist = (
322+
'group',
323+
'id',
324+
'ip_protocol',
325+
'ip_range',
326+
'parent_group_id',
327+
'port_range',
328+
)
329+
self.assertEqual(collist, columns)
330+
datalist = (
331+
{},
332+
security_group_rule_id,
333+
'icmp',
334+
'',
335+
security_group_id,
336+
'',
337+
)
338+
self.assertEqual(datalist, data)

0 commit comments

Comments
 (0)