Merge "Add security group extensions to OSC"

setup.cfg

@@ -53,6 +53,8 @@ openstack.nsxclient.v2 =
     port_set = vmware_nsx.osc.v2.port:NsxSetPort
     router_create = vmware_nsx.osc.v2.router:NsxCreateRouter
     router_set = vmware_nsx.osc.v2.router:NsxSetRouter
+    security_group_create = vmware_nsx.osc.v2.security_group:NsxCreateSecurityGroup
+    security_group_set = vmware_nsx.osc.v2.security_group:NsxSetSecurityGroup
     subnet_create = vmware_nsx.osc.v2.subnet:NsxCreateSubnet
     subnet_set = vmware_nsx.osc.v2.subnet:NsxSetSubnet
 

vmware_nsx/osc/v2/security_group.py

@@ -0,0 +1,137 @@
+# Copyright 2016 VMware, Inc.
+# All rights reserved.
+#
+#    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.
+
+"""Security group action implementations with nsx extensions"""
+
+from osc_lib import utils as osc_utils
+
+from openstackclient.identity import common as identity_common
+from openstackclient.network.v2 import security_group
+
+from vmware_nsx._i18n import _
+from vmware_nsx.osc.v2 import utils
+
+
+def add_nsx_extensions_to_parser(parser, client_manager, for_create=True):
+    if 'security-group-logging' in utils.get_extensions(client_manager):
+        # logging
+        logging_enable_group = parser.add_mutually_exclusive_group()
+        logging_enable_group.add_argument(
+            '--logging',
+            action='store_true',
+            help=_("Enable logging")
+        )
+        logging_enable_group.add_argument(
+            '--no-logging',
+            action='store_true',
+            help=_("Disable logging (default)")
+        )
+    if ('provider-security-group' in utils.get_extensions(client_manager)
+        and for_create):
+        # provider
+        parser.add_argument(
+            '--provider',
+            action='store_true',
+            help=_("Provider security group")
+        )
+
+
+def _get_plugin_attrs(attrs, parsed_args, client_manager):
+    if 'security-group-logging' in utils.get_extensions(client_manager):
+        # logging
+        if parsed_args.logging:
+            attrs['logging'] = True
+        if parsed_args.no_logging:
+            attrs['logging'] = False
+    if 'provider-security-group' in utils.get_extensions(client_manager):
+        # provider
+        if hasattr(parsed_args, 'provider') and parsed_args.provider:
+            attrs['provider'] = True
+
+    return attrs
+
+
+class NsxCreateSecurityGroup(security_group.CreateSecurityGroup):
+    """Create a new security_group with vmware nsx extensions """
+
+    def take_action_network(self, client, parsed_args):
+        #TODO(asarfaty): Better to change the neutron client code of
+        # CreateSecurityGroup:take_action_network to use an internal
+        # get_attributes, and override only this
+
+        # Build the create attributes.
+        attrs = {}
+        attrs['name'] = parsed_args.name
+        attrs['description'] = self._get_description(parsed_args)
+        if parsed_args.project is not None:
+            identity_client = self.app.client_manager.identity
+            project_id = identity_common.find_project(
+                identity_client,
+                parsed_args.project,
+                parsed_args.project_domain,
+            ).id
+            attrs['tenant_id'] = project_id
+
+        # add the plugin attributes
+        attrs = _get_plugin_attrs(attrs, parsed_args, self.app.client_manager)
+
+        # Create the security group and display the results.
+        obj = client.create_security_group(**attrs)
+        display_columns, property_columns = security_group._get_columns(obj)
+        data = osc_utils.get_item_properties(
+            obj,
+            property_columns,
+            formatters=security_group._formatters_network
+        )
+        return (display_columns, data)
+
+    def update_parser_common(self, parser):
+        parser = super(NsxCreateSecurityGroup, self).update_parser_common(
+            parser)
+
+        # Add the nsx attributes to the neutron security group attributes
+        add_nsx_extensions_to_parser(
+            parser, self.app.client_manager, for_create=True)
+        return parser
+
+
+class NsxSetSecurityGroup(security_group.SetSecurityGroup):
+    """Set security group properties with vmware nsx extensions """
+
+    def take_action_network(self, client, parsed_args):
+        #TODO(asarfaty): Better to change the neutron client code of
+        # CreateSecurityGroup:take_action_network to use an internal
+        # get_attributes, and override only this
+
+        obj = client.find_security_group(parsed_args.group,
+                                         ignore_missing=False)
+        attrs = {}
+        if parsed_args.name is not None:
+            attrs['name'] = parsed_args.name
+        if parsed_args.description is not None:
+            attrs['description'] = parsed_args.description
+
+        # add the plugin attributes
+        attrs = _get_plugin_attrs(attrs, parsed_args, self.app.client_manager)
+
+        client.update_security_group(obj, **attrs)
+
+    def update_parser_common(self, parser):
+        parser = super(NsxSetSecurityGroup, self).update_parser_common(parser)
+
+        # Add the nsx attributes to the neutron security group attributes
+        add_nsx_extensions_to_parser(
+            parser, self.app.client_manager, for_create=False)
+        return parser

vmware_nsx/tests/unit/osc/v2/test_security_group.py

@@ -0,0 +1,128 @@
+# Copyright 2016 VMware, Inc.
+# All Rights Reserved
+#
+#    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 mock
+import re
+
+from openstackclient.tests.unit.network.v2 import test_security_group
+from openstackclient.tests.unit import utils as tests_utils
+
+from vmware_nsx.osc.v2 import security_group
+
+
+supported_extensions = ('security-group-logging', 'provider-security-group')
+
+
+class TestCreateSecurityGroup(
+    test_security_group.TestCreateSecurityGroupNetwork):
+
+    def setUp(self):
+        super(TestCreateSecurityGroup, self).setUp()
+        # Get the command object to test
+        self.cmd = security_group.NsxCreateSecurityGroup(
+            self.app, self.namespace)
+        # mock the relevant extensions
+        get_ext = mock.patch('vmware_nsx.osc.v2.utils.get_extensions').start()
+        get_ext.return_value = supported_extensions
+
+    def _test_create_with_flag_arg(
+        self, arg_name, validate_name, validate_val):
+        self.network.create_security_group = mock.Mock(
+            return_value=self._security_group)
+        # add '--' to the arg name and change '_' to '-'
+        conv_name = '--' + re.sub('_', '-', arg_name)
+        arglist = [
+            '--description', self._security_group.description,
+            conv_name,
+            self._security_group.name
+        ]
+        verifylist = [
+            ('description', self._security_group.description),
+            ('name', self._security_group.name),
+            (arg_name, True),
+        ]
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        columns, data = self.cmd.take_action(parsed_args)
+
+        self.network.create_security_group.assert_called_once_with(**{
+            'description': self._security_group.description,
+            'name': self._security_group.name,
+            validate_name: validate_val,
+        })
+        self.assertEqual(self.columns, columns)
+        self.assertEqual(self.data, data)
+
+    def test_create_with_logging(self):
+        self._test_create_with_flag_arg('logging', 'logging', True)
+
+    def test_create_with_no_logging(self):
+        self._test_create_with_flag_arg('no_logging', 'logging', False)
+
+    def test_create_with_provider(self):
+        self._test_create_with_flag_arg('provider', 'provider', True)
+
+
+class TestSetSecurityGroup(test_security_group.TestSetSecurityGroupNetwork):
+
+    def setUp(self):
+        super(TestSetSecurityGroup, self).setUp()
+        # Get the command object to test
+        self.cmd = security_group.NsxSetSecurityGroup(
+            self.app, self.namespace)
+        # mock the relevant extensions
+        get_ext = mock.patch('vmware_nsx.osc.v2.utils.get_extensions').start()
+        get_ext.return_value = supported_extensions
+
+    def _test_set_with_flag_arg(self, arg_name, validate_name,
+                                validate_val, is_valid=True):
+        self.network.create_security_group = mock.Mock(
+            return_value=self._security_group)
+        # add '--' to the arg name and change '_' to '-'
+        conv_name = '--' + re.sub('_', '-', arg_name)
+        arglist = [
+            conv_name,
+            self._security_group.name
+        ]
+        verifylist = [
+            (arg_name, True),
+            ('group', self._security_group.name),
+        ]
+
+        if not is_valid:
+            self.assertRaises(tests_utils.ParserException, self.check_parser,
+                              self.cmd, arglist, verifylist)
+            return
+
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        result = self.cmd.take_action(parsed_args)
+
+        self.network.update_security_group.assert_called_once_with(
+            self._security_group,
+            **{validate_name: validate_val})
+        self.assertIsNone(result)
+
+    def test_set_with_logging(self):
+        self._test_set_with_flag_arg('logging', 'logging', True)
+
+    def test_set_with_no_logging(self):
+        self._test_set_with_flag_arg('no_logging', 'logging', False)
+
+    def test_set_with_provider(self):
+        # modifying the provider flag should fail
+        self._test_set_with_flag_arg('provider', 'provider',
+                                     True, is_valid=False)