openstack/python-openstackclient
Code Issues Proposed changes

Merge "Refactor security group show to use SDK"

Browse Source
This commit is contained in:
Jenkins
2016-03-11 01:33:50 +00:00
committed by Gerrit Code Review
parent 4bb48c088d 564c8ff240
commit d4b71ea890
9 changed files with 310 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
openstackclient/compute/v2/security_group.py
View File

@@ -56,23 +56,6 @@ def _xform_security_group_rule(sgroup):
return info
def _xform_and_trim_security_group_rule(sgroup):
info = _xform_security_group_rule(sgroup)
# Trim parent security group ID since caller has this information.
info.pop('parent_group_id', None)
# Trim keys with empty string values.
keys_to_trim = [
'ip_protocol',
'ip_range',
'port_range',
'remote_security_group',
]
for key in keys_to_trim:
if key in info and not info[key]:
info.pop(key)
return info
class CreateSecurityGroup(command.ShowOne):
"""Create a new security group"""
@@ -215,40 +198,3 @@ class ListSecurityGroupRule(command.Lister):
(utils.get_item_properties(
s, columns,
) for s in rules))
class ShowSecurityGroup(command.ShowOne):
"""Display security group details"""
def get_parser(self, prog_name):
parser = super(ShowSecurityGroup, self).get_parser(prog_name)
parser.add_argument(
'group',
metavar='<group>',
help='Security group to display (name or ID)',
)
return parser
def take_action(self, parsed_args):
compute_client = self.app.client_manager.compute
info = {}
info.update(utils.find_resource(
compute_client.security_groups,
parsed_args.group,
)._info)
rules = []
for r in info['rules']:
formatted_rule = _xform_and_trim_security_group_rule(r)
rules.append(utils.format_dict(formatted_rule))
# Format rules into a list of strings
info.update(
{'rules': utils.format_list(rules, separator='\n')}
)
# Map 'tenant_id' column to 'project_id'
info.update(
{'project_id': info.pop('tenant_id')}
)
return zip(*sorted(six.iteritems(info)))

41
openstackclient/network/utils.py Normal file
View File

@@ -0,0 +1,41 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
# Transform compute security group rule for display.
def transform_compute_security_group_rule(sg_rule):
info = {}
info.update(sg_rule)
from_port = info.pop('from_port')
to_port = info.pop('to_port')
if isinstance(from_port, int) and isinstance(to_port, int):
port_range = {'port_range': "%u:%u" % (from_port, to_port)}
elif from_port is None and to_port is None:
port_range = {'port_range': ""}
else:
port_range = {'port_range': "%s:%s" % (from_port, to_port)}
info.update(port_range)
if 'cidr' in info['ip_range']:
info['ip_range'] = info['ip_range']['cidr']
else:
info['ip_range'] = ''
if info['ip_protocol'] is None:
info['ip_protocol'] = ''
elif info['ip_protocol'].lower() == 'icmp':
info['port_range'] = ''
group = info.pop('group')
if 'name' in group:
info['remote_security_group'] = group['name']
else:
info['remote_security_group'] = ''
return info

108
openstackclient/network/v2/security_group.py
View File

@@ -14,9 +14,81 @@
"""Security Group action implementations"""
import argparse
import six
from openstackclient.common import utils
from openstackclient.network import common
from openstackclient.network import utils as network_utils
def _format_network_security_group_rules(sg_rules):
# For readability and to align with formatting compute security group
# rules, trim keys with caller known (e.g. security group and tenant ID)
# or empty values.
for sg_rule in sg_rules:
empty_keys = [k for k, v in six.iteritems(sg_rule) if not v]
for key in empty_keys:
sg_rule.pop(key)
sg_rule.pop('security_group_id', None)
sg_rule.pop('tenant_id', None)
return utils.format_list_of_dicts(sg_rules)
def _format_compute_security_group_rule(sg_rule):
info = network_utils.transform_compute_security_group_rule(sg_rule)
# Trim parent security group ID since caller has this information.
info.pop('parent_group_id', None)
# Trim keys with empty string values.
keys_to_trim = [
'ip_protocol',
'ip_range',
'port_range',
'remote_security_group',
]
for key in keys_to_trim:
if key in info and not info[key]:
info.pop(key)
return utils.format_dict(info)
def _format_compute_security_group_rules(sg_rules):
rules = []
for sg_rule in sg_rules:
rules.append(_format_compute_security_group_rule(sg_rule))
return utils.format_list(rules, separator='\n')
_formatters_network = {
'security_group_rules': _format_network_security_group_rules,
}
_formatters_compute = {
'rules': _format_compute_security_group_rules,
}
def _get_columns(item):
# Build the display columns and a list of the property columns
# that need to be mapped (display column name, property name).
columns = list(item.keys())
property_column_mappings = []
if 'security_group_rules' in columns:
columns.append('rules')
columns.remove('security_group_rules')
property_column_mappings.append(('rules', 'security_group_rules'))
if 'tenant_id' in columns:
columns.append('project_id')
columns.remove('tenant_id')
property_column_mappings.append(('project_id', 'tenant_id'))
display_columns = sorted(columns)
# Build the property columns and apply any column mappings.
property_columns = sorted(columns)
for property_column_mapping in property_column_mappings:
property_index = property_columns.index(property_column_mapping[0])
property_columns[property_index] = property_column_mapping[1]
return tuple(display_columns), property_columns
class DeleteSecurityGroup(common.NetworkAndComputeCommand):
@@ -143,3 +215,39 @@ class SetSecurityGroup(common.NetworkAndComputeCommand):
data.name,
data.description,
)
class ShowSecurityGroup(common.NetworkAndComputeShowOne):
"""Display security group details"""
def update_parser_common(self, parser):
parser.add_argument(
'group',
metavar='<group>',
help='Security group to display (name or ID)',
)
return parser
def take_action_network(self, client, parsed_args):
obj = client.find_security_group(parsed_args.group,
ignore_missing=False)
display_columns, property_columns = _get_columns(obj)
data = utils.get_item_properties(
obj,
property_columns,
formatters=_formatters_network
)
return (display_columns, data)
def take_action_compute(self, client, parsed_args):
obj = utils.find_resource(
client.security_groups,
parsed_args.group,
)
display_columns, property_columns = _get_columns(obj._info)
data = utils.get_dict_properties(
obj._info,
property_columns,
formatters=_formatters_compute
)
return (display_columns, data)

31
openstackclient/network/v2/security_group_rule.py
View File

@@ -18,38 +18,11 @@ import six
from openstackclient.common import exceptions
from openstackclient.common import utils
from openstackclient.network import common
def _xform_security_group_rule(sgroup):
info = {}
info.update(sgroup)
from_port = info.pop('from_port')
to_port = info.pop('to_port')
if isinstance(from_port, int) and isinstance(to_port, int):
port_range = {'port_range': "%u:%u" % (from_port, to_port)}
elif from_port is None and to_port is None:
port_range = {'port_range': ""}
else:
port_range = {'port_range': "%s:%s" % (from_port, to_port)}
info.update(port_range)
if 'cidr' in info['ip_range']:
info['ip_range'] = info['ip_range']['cidr']
else:
info['ip_range'] = ''
if info['ip_protocol'] is None:
info['ip_protocol'] = ''
elif info['ip_protocol'].lower() == 'icmp':
info['port_range'] = ''
group = info.pop('group')
if 'name' in group:
info['remote_security_group'] = group['name']
else:
info['remote_security_group'] = ''
return info
from openstackclient.network import utils as network_utils
def _format_security_group_rule_show(obj):
data = _xform_security_group_rule(obj)
data = network_utils.transform_compute_security_group_rule(obj)
return zip(*sorted(six.iteritems(data)))

13
openstackclient/tests/compute/v2/fakes.py
View File

@@ -333,7 +333,9 @@ class FakeSecurityGroup(object):
security_group_attrs.update(attrs)
# Set default methods.
security_group_methods = {}
security_group_methods = {
'keys': ['id', 'name', 'description', 'tenant_id', 'rules'],
}
# Overwrite default methods.
security_group_methods.update(methods)
@@ -369,7 +371,7 @@ class FakeSecurityGroupRule(object):
"""Fake one or more security group rules."""
@staticmethod
def create_one_security_group_rule(attrs={}, methods={}):
def create_one_security_group_rule(attrs=None, methods=None):
"""Create a fake security group rule.
:param Dictionary attrs:
@@ -379,6 +381,11 @@ class FakeSecurityGroupRule(object):
:return:
A FakeResource object, with id, etc.
"""
if attrs is None:
attrs = {}
if methods is None:
methods = {}
# Set default attributes.
security_group_rule_attrs = {
'from_port': -1,
@@ -406,7 +413,7 @@ class FakeSecurityGroupRule(object):
return security_group_rule
@staticmethod
def create_security_group_rules(attrs={}, methods={}, count=2):
def create_security_group_rules(attrs=None, methods=None, count=2):
"""Create multiple fake security group rules.
:param Dictionary attrs:

29
openstackclient/tests/network/v2/fakes.py
View File

@@ -419,7 +419,7 @@ class FakeSecurityGroup(object):
"""Fake one or more security groups."""
@staticmethod
def create_one_security_group(attrs={}, methods={}):
def create_one_security_group(attrs=None, methods=None):
"""Create a fake security group.
:param Dictionary attrs:
@@ -429,6 +429,11 @@ class FakeSecurityGroup(object):
:return:
A FakeResource object, with id, name, etc.
"""
if attrs is None:
attrs = {}
if methods is None:
methods = {}
# Set default attributes.
security_group_attrs = {
'id': 'security-group-id-' + uuid.uuid4().hex,
@@ -442,7 +447,10 @@ class FakeSecurityGroup(object):
security_group_attrs.update(attrs)
# Set default methods.
security_group_methods = {}
security_group_methods = {
'keys': ['id', 'name', 'description', 'tenant_id',
'security_group_rules'],
}
# Overwrite default methods.
security_group_methods.update(methods)
@@ -451,10 +459,14 @@ class FakeSecurityGroup(object):
info=copy.deepcopy(security_group_attrs),
methods=copy.deepcopy(security_group_methods),
loaded=True)
# Set attributes with special mapping in OpenStack SDK.
security_group.project_id = security_group_attrs['tenant_id']
return security_group
@staticmethod
def create_security_groups(attrs={}, methods={}, count=2):
def create_security_groups(attrs=None, methods=None, count=2):
"""Create multiple fake security groups.
:param Dictionary attrs:
@@ -478,7 +490,7 @@ class FakeSecurityGroupRule(object):
"""Fake one or more security group rules."""
@staticmethod
def create_one_security_group_rule(attrs={}, methods={}):
def create_one_security_group_rule(attrs=None, methods=None):
"""Create a fake security group rule.
:param Dictionary attrs:
@@ -488,6 +500,11 @@ class FakeSecurityGroupRule(object):
:return:
A FakeResource object, with id, etc.
"""
if attrs is None:
attrs = {}
if methods is None:
methods = {}
# Set default attributes.
security_group_rule_attrs = {
'direction': 'ingress',
@@ -520,13 +537,13 @@ class FakeSecurityGroupRule(object):
methods=copy.deepcopy(security_group_rule_methods),
loaded=True)
# Set attributes with special mappings.
# Set attributes with special mapping in OpenStack SDK.
security_group_rule.project_id = security_group_rule_attrs['tenant_id']
return security_group_rule
@staticmethod
def create_security_group_rules(attrs={}, methods={}, count=2):
def create_security_group_rules(attrs=None, methods=None, count=2):
"""Create multiple fake security group rules.
:param Dictionary attrs:

119
openstackclient/tests/network/v2/test_security_group.py
View File

@@ -363,3 +363,122 @@ class TestSetSecurityGroupCompute(TestSecurityGroupCompute):
new_description
)
self.assertIsNone(result)
class TestShowSecurityGroupNetwork(TestSecurityGroupNetwork):
# The security group rule to be shown with the group.
_security_group_rule = \
network_fakes.FakeSecurityGroupRule.create_one_security_group_rule()
# The security group to be shown.
_security_group = \
network_fakes.FakeSecurityGroup.create_one_security_group(
attrs={'security_group_rules': [_security_group_rule._info]}
)
columns = (
'description',
'id',
'name',
'project_id',
'rules',
)
data = (
_security_group.description,
_security_group.id,
_security_group.name,
_security_group.project_id,
security_group._format_network_security_group_rules(
[_security_group_rule._info]),
)
def setUp(self):
super(TestShowSecurityGroupNetwork, self).setUp()
self.network.find_security_group = mock.Mock(
return_value=self._security_group)
# Get the command object to test
self.cmd = security_group.ShowSecurityGroup(self.app, self.namespace)
def test_show_no_options(self):
self.assertRaises(tests_utils.ParserException,
self.check_parser, self.cmd, [], [])
def test_show_all_options(self):
arglist = [
self._security_group.id,
]
verifylist = [
('group', self._security_group.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.network.find_security_group.assert_called_once_with(
self._security_group.id, ignore_missing=False)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
class TestShowSecurityGroupCompute(TestSecurityGroupCompute):
# The security group rule to be shown with the group.
_security_group_rule = \
compute_fakes.FakeSecurityGroupRule.create_one_security_group_rule()
# The security group to be shown.
_security_group = \
compute_fakes.FakeSecurityGroup.create_one_security_group(
attrs={'rules': [_security_group_rule._info]}
)
columns = (
'description',
'id',
'name',
'project_id',
'rules',
)
data = (
_security_group.description,
_security_group.id,
_security_group.name,
_security_group.tenant_id,
security_group._format_compute_security_group_rules(
[_security_group_rule._info]),
)
def setUp(self):
super(TestShowSecurityGroupCompute, self).setUp()
self.app.client_manager.network_endpoint_enabled = False
self.compute.security_groups.get.return_value = self._security_group
# Get the command object to test
self.cmd = security_group.ShowSecurityGroup(self.app, None)
def test_show_no_options(self):
self.assertRaises(tests_utils.ParserException,
self.check_parser, self.cmd, [], [])
def test_show_all_options(self):
arglist = [
self._security_group.id,
]
verifylist = [
('group', self._security_group.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.compute.security_groups.get.assert_called_once_with(
self._security_group.id)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)

6
releasenotes/notes/bug-1519511-68ca30ad5519d3d8.yaml Normal file
View File

@@ -0,0 +1,6 @@
---
features:
- The ``security group show`` command now uses Network v2 when
enabled which results in a more detailed output for network
security group rules.
[Bug `1519511 <https://bugs.launchpad.net/bugs/1519511>`_]

2
setup.cfg
View File

@@ -100,7 +100,6 @@ openstack.compute.v2 =
keypair_show = openstackclient.compute.v2.keypair:ShowKeypair
security_group_create = openstackclient.compute.v2.security_group:CreateSecurityGroup
security_group_show = openstackclient.compute.v2.security_group:ShowSecurityGroup
security_group_rule_create = openstackclient.compute.v2.security_group:CreateSecurityGroupRule
security_group_rule_list = openstackclient.compute.v2.security_group:ListSecurityGroupRule
@@ -348,6 +347,7 @@ openstack.network.v2 =
security_group_delete = openstackclient.network.v2.security_group:DeleteSecurityGroup
security_group_list = openstackclient.network.v2.security_group:ListSecurityGroup
security_group_set = openstackclient.network.v2.security_group:SetSecurityGroup
security_group_show = openstackclient.network.v2.security_group:ShowSecurityGroup
security_group_rule_delete = openstackclient.network.v2.security_group_rule:DeleteSecurityGroupRule
security_group_rule_show = openstackclient.network.v2.security_group_rule:ShowSecurityGroupRule