Merge "Refactor security group rule list to use SDK"

4 files changed, 301 insertions, 332 deletions

diff --git a/openstackclient/compute/v2/security_group.py b/openstackclient/compute/v2/security_group.py
deleted file mode 100644
index ca3bf5dc..00000000
--- a/openstackclient/compute/v2/security_group.py
+++ /dev/null
@@ -1,103 +0,0 @@
-#   Copyright 2012 OpenStack Foundation
-#   Copyright 2013 Nebula Inc
-#
-#   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.
-#
-
-"""Compute v2 Security Group action implementations"""
-
-try:
-    from novaclient.v2 import security_group_rules
-except ImportError:
-    from novaclient.v1_1 import security_group_rules
-
-from openstackclient.common import command
-from openstackclient.common import utils
-
-
-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
-
-
-class ListSecurityGroupRule(command.Lister):
-    """List security group rules"""
-
-    def get_parser(self, prog_name):
-        parser = super(ListSecurityGroupRule, self).get_parser(prog_name)
-        parser.add_argument(
-            'group',
-            metavar='<group>',
-            nargs='?',
-            help='List all rules in this security group (name or ID)',
-        )
-        return parser
-
-    def take_action(self, parsed_args):
-        compute_client = self.app.client_manager.compute
-        columns = column_headers = (
-            "ID",
-            "IP Protocol",
-            "IP Range",
-            "Port Range",
-            "Remote Security Group",
-        )
-
-        rules_to_list = []
-        if parsed_args.group:
-            group = utils.find_resource(
-                compute_client.security_groups,
-                parsed_args.group,
-            )
-            rules_to_list = group.rules
-        else:
-            columns = columns + ('parent_group_id',)
-            column_headers = column_headers + ('Security Group',)
-            for group in compute_client.security_groups.list():
-                rules_to_list.extend(group.rules)
-
-        # Argh, the rules are not Resources...
-        rules = []
-        for rule in rules_to_list:
-            rules.append(security_group_rules.SecurityGroupRule(
-                compute_client.security_group_rules,
-                _xform_security_group_rule(rule),
-            ))
-
-        return (column_headers,
-                (utils.get_item_properties(
-                    s, columns,
-                ) for s in rules))
diff --git a/openstackclient/network/v2/security_group_rule.py b/openstackclient/network/v2/security_group_rule.py
index f60995ab..c6fb3558 100644
--- a/openstackclient/network/v2/security_group_rule.py
+++ b/openstackclient/network/v2/security_group_rule.py
@@ -15,6 +15,11 @@
 
 import six
 
+try:
+    from novaclient.v2 import security_group_rules as compute_secgroup_rules
+except ImportError:
+    from novaclient.v1_1 import security_group_rules as compute_secgroup_rules
+
 from openstackclient.common import exceptions
 from openstackclient.common import parseractions
 from openstackclient.common import utils
@@ -27,6 +32,20 @@ def _format_security_group_rule_show(obj):
     return zip(*sorted(six.iteritems(data)))
 
 
+def _format_network_port_range(rule):
+    port_range = ''
+    if (rule.protocol != 'icmp' and
+            (rule.port_range_min or rule.port_range_max)):
+        port_range_min = str(rule.port_range_min)
+        port_range_max = str(rule.port_range_max)
+        if rule.port_range_min is None:
+            port_range_min = port_range_max
+        if rule.port_range_max is None:
+            port_range_max = port_range_min
+        port_range = port_range_min + ':' + port_range_max
+    return port_range
+
+
 def _get_columns(item):
     columns = list(item.keys())
     if 'tenant_id' in columns:
@@ -161,6 +180,102 @@ class DeleteSecurityGroupRule(common.NetworkAndComputeCommand):
         client.security_group_rules.delete(parsed_args.rule)
 
 
+class ListSecurityGroupRule(common.NetworkAndComputeLister):
+    """List security group rules"""
+
+    def update_parser_common(self, parser):
+        parser.add_argument(
+            'group',
+            metavar='<group>',
+            nargs='?',
+            help='List all rules in this security group (name or ID)',
+        )
+        return parser
+
+    def _get_column_headers(self, parsed_args):
+        column_headers = (
+            'ID',
+            'IP Protocol',
+            'IP Range',
+            'Port Range',
+            'Remote Security Group',
+        )
+        if parsed_args.group is None:
+            column_headers = column_headers + ('Security Group',)
+        return column_headers
+
+    def take_action_network(self, client, parsed_args):
+        column_headers = self._get_column_headers(parsed_args)
+        columns = (
+            'id',
+            'protocol',
+            'remote_ip_prefix',
+            'port_range_min',
+            'remote_group_id',
+        )
+
+        # Get the security group rules using the requested query.
+        query = {}
+        if parsed_args.group is not None:
+            # NOTE(rtheis): Unfortunately, the security group resource
+            # does not contain security group rules resources. So use
+            # the security group ID in a query to get the resources.
+            security_group_id = client.find_security_group(
+                parsed_args.group,
+                ignore_missing=False
+            ).id
+            query = {'security_group_id': security_group_id}
+        else:
+            columns = columns + ('security_group_id',)
+        rules = list(client.security_group_rules(**query))
+
+        # Reformat the rules to display a port range instead
+        # of just the port range minimum. This maintains
+        # output compatibility with compute.
+        for rule in rules:
+            rule.port_range_min = _format_network_port_range(rule)
+
+        return (column_headers,
+                (utils.get_item_properties(
+                    s, columns,
+                ) for s in rules))
+
+    def take_action_compute(self, client, parsed_args):
+        column_headers = self._get_column_headers(parsed_args)
+        columns = (
+            "ID",
+            "IP Protocol",
+            "IP Range",
+            "Port Range",
+            "Remote Security Group",
+        )
+
+        rules_to_list = []
+        if parsed_args.group is not None:
+            group = utils.find_resource(
+                client.security_groups,
+                parsed_args.group,
+            )
+            rules_to_list = group.rules
+        else:
+            columns = columns + ('parent_group_id',)
+            for group in client.security_groups.list():
+                rules_to_list.extend(group.rules)
+
+        # NOTE(rtheis): Turn the raw rules into resources.
+        rules = []
+        for rule in rules_to_list:
+            rules.append(compute_secgroup_rules.SecurityGroupRule(
+                client.security_group_rules,
+                network_utils.transform_compute_security_group_rule(rule),
+            ))
+
+        return (column_headers,
+                (utils.get_item_properties(
+                    s, columns,
+                ) for s in rules))
+
+
 class ShowSecurityGroupRule(common.NetworkAndComputeShowOne):
     """Display security group rule details"""
 
diff --git a/openstackclient/tests/compute/v2/test_security_group_rule.py b/openstackclient/tests/compute/v2/test_security_group_rule.py
deleted file mode 100644
index 42bf2c26..00000000
--- a/openstackclient/tests/compute/v2/test_security_group_rule.py
+++ /dev/null
@@ -1,229 +0,0 @@
-#   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.
-#
-
-import copy
-
-from openstackclient.compute.v2 import security_group
-from openstackclient.tests.compute.v2 import fakes as compute_fakes
-from openstackclient.tests import fakes
-from openstackclient.tests.identity.v2_0 import fakes as identity_fakes
-
-
-security_group_id = '11'
-security_group_name = 'wide-open'
-security_group_description = 'nothing but net'
-
-security_group_rule_id = '1'
-security_group_rule_cidr = '0.0.0.0/0'
-
-SECURITY_GROUP_RULE = {
-    'id': security_group_rule_id,
-    'group': {},
-    'ip_protocol': 'tcp',
-    'ip_range': {'cidr': security_group_rule_cidr},
-    'parent_group_id': security_group_id,
-    'from_port': 0,
-    'to_port': 0,
-}
-
-SECURITY_GROUP_RULE_ICMP = {
-    'id': security_group_rule_id,
-    'group': {},
-    'ip_protocol': 'icmp',
-    'ip_range': {'cidr': security_group_rule_cidr},
-    'parent_group_id': security_group_id,
-    'from_port': -1,
-    'to_port': -1,
-}
-
-SECURITY_GROUP_RULE_REMOTE_GROUP = {
-    'id': security_group_rule_id,
-    'group': {"tenant_id": "14", "name": "default"},
-    'ip_protocol': 'tcp',
-    'ip_range': {},
-    'parent_group_id': security_group_id,
-    'from_port': 80,
-    'to_port': 80,
-}
-
-SECURITY_GROUP = {
-    'id': security_group_id,
-    'name': security_group_name,
-    'description': security_group_description,
-    'tenant_id': identity_fakes.project_id,
-    'rules': [SECURITY_GROUP_RULE,
-              SECURITY_GROUP_RULE_ICMP,
-              SECURITY_GROUP_RULE_REMOTE_GROUP],
-}
-
-security_group_2_id = '12'
-security_group_2_name = 'he-shoots'
-security_group_2_description = 'he scores'
-
-SECURITY_GROUP_2_RULE = {
-    'id': '2',
-    'group': {},
-    'ip_protocol': 'tcp',
-    'ip_range': {},
-    'parent_group_id': security_group_2_id,
-    'from_port': 80,
-    'to_port': 80,
-}
-
-SECURITY_GROUP_2 = {
-    'id': security_group_2_id,
-    'name': security_group_2_name,
-    'description': security_group_2_description,
-    'tenant_id': identity_fakes.project_id,
-    'rules': [SECURITY_GROUP_2_RULE],
-}
-
-
-class FakeSecurityGroupRuleResource(fakes.FakeResource):
-
-    def get_keys(self):
-        return {'property': 'value'}
-
-
-class TestSecurityGroupRule(compute_fakes.TestComputev2):
-
-    def setUp(self):
-        super(TestSecurityGroupRule, self).setUp()
-
-        # Get a shortcut compute client security_groups mock
-        self.secgroups_mock = self.app.client_manager.compute.security_groups
-        self.secgroups_mock.reset_mock()
-
-        # Get a shortcut compute client security_group_rules mock
-        self.sg_rules_mock = \
-            self.app.client_manager.compute.security_group_rules
-        self.sg_rules_mock.reset_mock()
-
-
-class TestSecurityGroupRuleList(TestSecurityGroupRule):
-
-    def setUp(self):
-        super(TestSecurityGroupRuleList, self).setUp()
-
-        security_group_mock = FakeSecurityGroupRuleResource(
-            None,
-            copy.deepcopy(SECURITY_GROUP),
-            loaded=True,
-        )
-
-        security_group_2_mock = FakeSecurityGroupRuleResource(
-            None,
-            copy.deepcopy(SECURITY_GROUP_2),
-            loaded=True,
-        )
-
-        self.secgroups_mock.get.return_value = security_group_mock
-        self.secgroups_mock.list.return_value = [security_group_mock,
-                                                 security_group_2_mock]
-
-        # Get the command object to test
-        self.cmd = security_group.ListSecurityGroupRule(self.app, None)
-
-    def test_security_group_rule_list(self):
-
-        arglist = [
-            security_group_name,
-        ]
-        verifylist = [
-            ('group', security_group_name),
-        ]
-
-        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
-
-        # In base command class Lister in cliff, abstract method take_action()
-        # returns a tuple containing the column names and an iterable
-        # containing the data to be listed.
-        columns, data = self.cmd.take_action(parsed_args)
-
-        collist = (
-            'ID',
-            'IP Protocol',
-            'IP Range',
-            'Port Range',
-            'Remote Security Group',
-        )
-        self.assertEqual(collist, columns)
-        datalist = ((
-            security_group_rule_id,
-            'tcp',
-            security_group_rule_cidr,
-            '0:0',
-            '',
-        ), (
-            security_group_rule_id,
-            'icmp',
-            security_group_rule_cidr,
-            '',
-            '',
-        ), (
-            security_group_rule_id,
-            'tcp',
-            '',
-            '80:80',
-            'default',
-        ),)
-        self.assertEqual(datalist, tuple(data))
-
-    def test_security_group_rule_list_no_group(self):
-
-        parsed_args = self.check_parser(self.cmd, [], [])
-
-        # In base command class Lister in cliff, abstract method take_action()
-        # returns a tuple containing the column names and an iterable
-        # containing the data to be listed.
-        columns, data = self.cmd.take_action(parsed_args)
-
-        collist = (
-            'ID',
-            'IP Protocol',
-            'IP Range',
-            'Port Range',
-            'Remote Security Group',
-            'Security Group',
-        )
-        self.assertEqual(collist, columns)
-        datalist = ((
-            security_group_rule_id,
-            'tcp',
-            security_group_rule_cidr,
-            '0:0',
-            '',
-            security_group_id,
-        ), (
-            security_group_rule_id,
-            'icmp',
-            security_group_rule_cidr,
-            '',
-            '',
-            security_group_id,
-        ), (
-            security_group_rule_id,
-            'tcp',
-            '',
-            '80:80',
-            'default',
-            security_group_id,
-        ), (
-            '2',
-            'tcp',
-            '',
-            '80:80',
-            '',
-            security_group_2_id,
-        ),)
-        self.assertEqual(datalist, tuple(data))
diff --git a/openstackclient/tests/network/v2/test_security_group_rule.py b/openstackclient/tests/network/v2/test_security_group_rule.py
index 81b9e18b..6aad8599 100644
--- a/openstackclient/tests/network/v2/test_security_group_rule.py
+++ b/openstackclient/tests/network/v2/test_security_group_rule.py
@@ -14,6 +14,7 @@
 import copy
 import mock
 
+from openstackclient.network import utils as network_utils
 from openstackclient.network.v2 import security_group_rule
 from openstackclient.tests.compute.v2 import fakes as compute_fakes
 from openstackclient.tests import fakes
@@ -414,6 +415,191 @@ class TestDeleteSecurityGroupRuleCompute(TestSecurityGroupRuleCompute):
         self.assertIsNone(result)
 
 
+class TestListSecurityGroupRuleNetwork(TestSecurityGroupRuleNetwork):
+
+    # The security group to hold the rules.
+    _security_group = \
+        network_fakes.FakeSecurityGroup.create_one_security_group()
+
+    # The security group rule to be listed.
+    _security_group_rule_tcp = \
+        network_fakes.FakeSecurityGroupRule.create_one_security_group_rule({
+            'protocol': 'tcp',
+            'port_range_max': 80,
+            'port_range_min': 80,
+            'security_group_id': _security_group.id,
+        })
+    _security_group_rule_icmp = \
+        network_fakes.FakeSecurityGroupRule.create_one_security_group_rule({
+            'protocol': 'icmp',
+            'port_range_max': -1,
+            'port_range_min': -1,
+            'remote_ip_prefix': '10.0.2.0/24',
+            'security_group_id': _security_group.id,
+        })
+    _security_group.security_group_rules = [_security_group_rule_tcp._info,
+                                            _security_group_rule_icmp._info]
+    _security_group_rules = [_security_group_rule_tcp,
+                             _security_group_rule_icmp]
+
+    expected_columns_with_group = (
+        'ID',
+        'IP Protocol',
+        'IP Range',
+        'Port Range',
+        'Remote Security Group',
+    )
+    expected_columns_no_group = \
+        expected_columns_with_group + ('Security Group',)
+
+    expected_data_with_group = []
+    expected_data_no_group = []
+    for _security_group_rule in _security_group_rules:
+        expected_rule_with_group = (
+            _security_group_rule.id,
+            _security_group_rule.protocol,
+            _security_group_rule.remote_ip_prefix,
+            security_group_rule._format_network_port_range(
+                _security_group_rule),
+            _security_group_rule.remote_group_id,
+        )
+        expected_rule_no_group = expected_rule_with_group + \
+            (_security_group_rule.security_group_id,)
+        expected_data_with_group.append(expected_rule_with_group)
+        expected_data_no_group.append(expected_rule_no_group)
+
+    def setUp(self):
+        super(TestListSecurityGroupRuleNetwork, self).setUp()
+
+        self.network.find_security_group = mock.Mock(
+            return_value=self._security_group)
+        self.network.security_group_rules = mock.Mock(
+            return_value=self._security_group_rules)
+
+        # Get the command object to test
+        self.cmd = security_group_rule.ListSecurityGroupRule(
+            self.app, self.namespace)
+
+    def test_list_no_group(self):
+        self._security_group_rule_tcp.port_range_min = 80
+        parsed_args = self.check_parser(self.cmd, [], [])
+
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.network.security_group_rules.assert_called_once_with(**{})
+        self.assertEqual(self.expected_columns_no_group, columns)
+        self.assertEqual(self.expected_data_no_group, list(data))
+
+    def test_list_with_group(self):
+        self._security_group_rule_tcp.port_range_min = 80
+        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.security_group_rules.assert_called_once_with(**{
+            'security_group_id': self._security_group.id,
+        })
+        self.assertEqual(self.expected_columns_with_group, columns)
+        self.assertEqual(self.expected_data_with_group, list(data))
+
+
+class TestListSecurityGroupRuleCompute(TestSecurityGroupRuleCompute):
+
+    # The security group to hold the rules.
+    _security_group = \
+        compute_fakes.FakeSecurityGroup.create_one_security_group()
+
+    # The security group rule to be listed.
+    _security_group_rule_tcp = \
+        compute_fakes.FakeSecurityGroupRule.create_one_security_group_rule({
+            'ip_protocol': 'tcp',
+            'from_port': 80,
+            'to_port': 80,
+            'group': {'name': _security_group.name},
+        })
+    _security_group_rule_icmp = \
+        compute_fakes.FakeSecurityGroupRule.create_one_security_group_rule({
+            'ip_protocol': 'icmp',
+            'from_port': -1,
+            'to_port': -1,
+            'ip_range': {'cidr': '10.0.2.0/24'},
+            'group': {'name': _security_group.name},
+        })
+    _security_group.rules = [_security_group_rule_tcp._info,
+                             _security_group_rule_icmp._info]
+
+    expected_columns_with_group = (
+        'ID',
+        'IP Protocol',
+        'IP Range',
+        'Port Range',
+        'Remote Security Group',
+    )
+    expected_columns_no_group = \
+        expected_columns_with_group + ('Security Group',)
+
+    expected_data_with_group = []
+    expected_data_no_group = []
+    for _security_group_rule in _security_group.rules:
+        rule = network_utils.transform_compute_security_group_rule(
+            _security_group_rule
+        )
+        expected_rule_with_group = (
+            rule['id'],
+            rule['ip_protocol'],
+            rule['ip_range'],
+            rule['port_range'],
+            rule['remote_security_group'],
+        )
+        expected_rule_no_group = expected_rule_with_group + \
+            (_security_group_rule['parent_group_id'],)
+        expected_data_with_group.append(expected_rule_with_group)
+        expected_data_no_group.append(expected_rule_no_group)
+
+    def setUp(self):
+        super(TestListSecurityGroupRuleCompute, self).setUp()
+
+        self.app.client_manager.network_endpoint_enabled = False
+
+        self.compute.security_groups.get.return_value = \
+            self._security_group
+        self.compute.security_groups.list.return_value = \
+            [self._security_group]
+
+        # Get the command object to test
+        self.cmd = security_group_rule.ListSecurityGroupRule(self.app, None)
+
+    def test_list_no_group(self):
+        parsed_args = self.check_parser(self.cmd, [], [])
+
+        columns, data = self.cmd.take_action(parsed_args)
+        self.compute.security_groups.list.assert_called_once_with()
+        self.assertEqual(self.expected_columns_no_group, columns)
+        self.assertEqual(self.expected_data_no_group, list(data))
+
+    def test_list_with_group(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.expected_columns_with_group, columns)
+        self.assertEqual(self.expected_data_with_group, list(data))
+
+
 class TestShowSecurityGroupRuleNetwork(TestSecurityGroupRuleNetwork):
 
     # The security group rule to be shown.