diff options
Diffstat (limited to 'openstackclient')
| -rw-r--r-- | openstackclient/compute/v2/security_group.py | 18 | ||||
| -rw-r--r-- | openstackclient/network/common.py | 8 | ||||
| -rw-r--r-- | openstackclient/network/v2/network.py | 45 | ||||
| -rw-r--r-- | openstackclient/network/v2/security_group_rule.py | 35 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/fakes.py | 149 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/fakes.py | 74 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/test_network.py | 162 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/test_security_group_rule.py | 100 |
8 files changed, 547 insertions, 44 deletions
diff --git a/openstackclient/compute/v2/security_group.py b/openstackclient/compute/v2/security_group.py index 2a8908d7..6f2e1a52 100644 --- a/openstackclient/compute/v2/security_group.py +++ b/openstackclient/compute/v2/security_group.py @@ -169,24 +169,6 @@ class CreateSecurityGroupRule(command.ShowOne): return zip(*sorted(six.iteritems(info))) -class DeleteSecurityGroupRule(command.Command): - """Delete a security group rule""" - - def get_parser(self, prog_name): - parser = super(DeleteSecurityGroupRule, self).get_parser(prog_name) - parser.add_argument( - 'rule', - metavar='<rule>', - help='Security group rule to delete (ID only)', - ) - return parser - - def take_action(self, parsed_args): - - compute_client = self.app.client_manager.compute - compute_client.security_group_rules.delete(parsed_args.rule) - - class ListSecurityGroup(command.Lister): """List security groups""" diff --git a/openstackclient/network/common.py b/openstackclient/network/common.py index cc343c3c..1e2c4cce 100644 --- a/openstackclient/network/common.py +++ b/openstackclient/network/common.py @@ -19,7 +19,13 @@ from openstackclient.common import command @six.add_metaclass(abc.ABCMeta) class NetworkAndComputeCommand(command.Command): - """Network and Compute Command""" + """Network and Compute Command + + Command class for commands that support implementation via + the network or compute endpoint. Such commands have different + implementations for take_action() and may even have different + arguments. + """ def take_action(self, parsed_args): if self.app.client_manager.is_network_endpoint_enabled(): diff --git a/openstackclient/network/v2/network.py b/openstackclient/network/v2/network.py index 636c333e..a634378f 100644 --- a/openstackclient/network/v2/network.py +++ b/openstackclient/network/v2/network.py @@ -168,11 +168,10 @@ class DeleteNetwork(common.NetworkAndComputeCommand): client.networks.delete(network.id) -class ListNetwork(command.Lister): +class ListNetwork(common.NetworkAndComputeLister): """List networks""" - def get_parser(self, prog_name): - parser = super(ListNetwork, self).get_parser(prog_name) + def update_parser_common(self, parser): parser.add_argument( '--external', action='store_true', @@ -187,9 +186,7 @@ class ListNetwork(command.Lister): ) return parser - def take_action(self, parsed_args): - client = self.app.client_manager.network - + def take_action_network(self, client, parsed_args): if parsed_args.long: columns = ( 'id', @@ -231,7 +228,29 @@ class ListNetwork(command.Lister): args = {'router:external': True} else: args = {} + data = client.networks(**args) + + return (column_headers, + (utils.get_item_properties( + s, columns, + formatters=_formatters, + ) for s in data)) + + def take_action_compute(self, client, parsed_args): + columns = ( + 'id', + 'label', + 'cidr', + ) + column_headers = ( + 'ID', + 'Name', + 'Subnet', + ) + + data = client.networks.list() + return (column_headers, (utils.get_item_properties( s, columns, @@ -297,7 +316,7 @@ class SetNetwork(command.Command): return -class ShowNetwork(command.ShowOne): +class ShowNetwork(common.NetworkAndComputeShowOne): """Show network details""" def get_parser(self, prog_name): @@ -309,9 +328,17 @@ class ShowNetwork(command.ShowOne): ) return parser - def take_action(self, parsed_args): - client = self.app.client_manager.network + def take_action_network(self, client, parsed_args): obj = client.find_network(parsed_args.network, ignore_missing=False) columns = _get_columns(obj) data = utils.get_item_properties(obj, columns, formatters=_formatters) return (columns, data) + + def take_action_compute(self, client, parsed_args): + network = utils.find_resource( + client.networks, + parsed_args.network, + ) + columns = sorted(network._info.keys()) + data = utils.get_dict_properties(network._info, columns) + return (columns, data) diff --git a/openstackclient/network/v2/security_group_rule.py b/openstackclient/network/v2/security_group_rule.py new file mode 100644 index 00000000..beeeaff7 --- /dev/null +++ b/openstackclient/network/v2/security_group_rule.py @@ -0,0 +1,35 @@ +# 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 Rule action implementations""" + +from openstackclient.network import common + + +class DeleteSecurityGroupRule(common.NetworkAndComputeCommand): + """Delete a security group rule""" + + def update_parser_common(self, parser): + parser.add_argument( + 'rule', + metavar='<rule>', + help='Security group rule to delete (ID only)', + ) + return parser + + def take_action_network(self, client, parsed_args): + obj = client.find_security_group_rule(parsed_args.rule) + client.delete_security_group_rule(obj) + + def take_action_compute(self, client, parsed_args): + client.security_group_rules.delete(parsed_args.rule) diff --git a/openstackclient/tests/compute/v2/fakes.py b/openstackclient/tests/compute/v2/fakes.py index 00f73748..974095ea 100644 --- a/openstackclient/tests/compute/v2/fakes.py +++ b/openstackclient/tests/compute/v2/fakes.py @@ -235,6 +235,68 @@ class FakeHypervisor(object): return hypervisors +class FakeSecurityGroupRule(object): + """Fake one or more security group rules.""" + + @staticmethod + def create_one_security_group_rule(attrs={}, methods={}): + """Create a fake security group rule. + + :param Dictionary attrs: + A dictionary with all attributes + :param Dictionary methods: + A dictionary with all methods + :return: + A FakeResource object, with id, etc. + """ + # Set default attributes. + security_group_rule_attrs = { + 'from_port': -1, + 'group': {}, + 'id': 'security-group-rule-id-' + uuid.uuid4().hex, + 'ip_protocol': 'icmp', + 'ip_range': {'cidr': '0.0.0.0/0'}, + 'parent_group_id': 'security-group-id-' + uuid.uuid4().hex, + 'to_port': -1, + } + + # Overwrite default attributes. + security_group_rule_attrs.update(attrs) + + # Set default methods. + security_group_rule_methods = {} + + # Overwrite default methods. + security_group_rule_methods.update(methods) + + security_group_rule = fakes.FakeResource( + info=copy.deepcopy(security_group_rule_attrs), + methods=copy.deepcopy(security_group_rule_methods), + loaded=True) + return security_group_rule + + @staticmethod + def create_security_group_rules(attrs={}, methods={}, count=2): + """Create multiple fake security group rules. + + :param Dictionary attrs: + A dictionary with all attributes + :param Dictionary methods: + A dictionary with all methods + :param int count: + The number of security group rules to fake + :return: + A list of FakeResource objects faking the security group rules + """ + security_group_rules = [] + for i in range(0, count): + security_group_rules.append( + FakeSecurityGroupRule.create_one_security_group_rule( + attrs, methods)) + + return security_group_rules + + class FakeServer(object): """Fake one or more compute servers.""" @@ -525,3 +587,90 @@ class FakeFloatingIP(object): if floating_ips is None: floating_ips = FakeFloatingIP.create_floating_ips(count) return mock.MagicMock(side_effect=floating_ips) + + +class FakeNetwork(object): + """Fake one or more networks.""" + + @staticmethod + def create_one_network(attrs={}, methods={}): + """Create a fake network. + + :param Dictionary attrs: + A dictionary with all attributes + :param Dictionary methods: + A dictionary with all methods + :return: + A FakeResource object, with id, label, cidr and so on + """ + # Set default attributes. + network_attrs = { + 'bridge': 'br100', + 'bridge_interface': None, + 'broadcast': '10.0.0.255', + 'cidr': '10.0.0.0/24', + 'cidr_v6': None, + 'created_at': '2016-02-11T11:17:37.000000', + 'deleted': False, + 'deleted_at': None, + 'dhcp_server': '10.0.0.1', + 'dhcp_start': '10.0.0.2', + 'dns1': '8.8.4.4', + 'dns2': None, + 'enable_dhcp': True, + 'gateway': '10.0.0.1', + 'gateway_v6': None, + 'host': None, + 'id': 'network-id-' + uuid.uuid4().hex, + 'injected': False, + 'label': 'network-label-' + uuid.uuid4().hex, + 'mtu': None, + 'multi_host': False, + 'netmask': '255.255.255.0', + 'netmask_v6': None, + 'priority': None, + 'project_id': 'project-id-' + uuid.uuid4().hex, + 'rxtx_base': None, + 'share_address': False, + 'updated_at': None, + 'vlan': None, + 'vpn_private_address': None, + 'vpn_public_address': None, + 'vpn_public_port': None, + } + + # Overwrite default attributes. + network_attrs.update(attrs) + + # Set default methods. + network_methods = { + 'keys': ['id', 'label', 'cidr'], + } + + # Overwrite default methods. + network_methods.update(methods) + + network = fakes.FakeResource(info=copy.deepcopy(network_attrs), + methods=copy.deepcopy(network_methods), + loaded=True) + + return network + + @staticmethod + def create_networks(attrs={}, methods={}, count=2): + """Create multiple fake networks. + + :param Dictionary attrs: + A dictionary with all attributes + :param Dictionary methods: + A dictionary with all methods + :param int count: + The number of networks to fake + :return: + A list of FakeResource objects faking the networks + """ + networks = [] + for i in range(0, count): + networks.append(FakeNetwork.create_one_network(attrs, methods)) + + return networks diff --git a/openstackclient/tests/network/v2/fakes.py b/openstackclient/tests/network/v2/fakes.py index aa04b1c0..b48cde3e 100644 --- a/openstackclient/tests/network/v2/fakes.py +++ b/openstackclient/tests/network/v2/fakes.py @@ -462,28 +462,76 @@ class FakeSecurityGroup(object): security_groups = [] for i in range(0, count): security_groups.append( - FakeRouter.create_one_security_group(attrs, methods)) + FakeSecurityGroup.create_one_security_group(attrs, methods)) return security_groups + +class FakeSecurityGroupRule(object): + """Fake one or more security group rules.""" + @staticmethod - def get_security_groups(security_groups=None, count=2): - """Get an iterable MagicMock object with a list of faked security groups. + def create_one_security_group_rule(attrs={}, methods={}): + """Create a fake security group rule. - If security group list is provided, then initialize the Mock object - with the list. Otherwise create one. + :param Dictionary attrs: + A dictionary with all attributes + :param Dictionary methods: + A dictionary with all methods + :return: + A FakeResource object, with id, name, etc. + """ + # Set default attributes. + security_group_rule_attrs = { + 'description': 'security-group-rule-desc-' + uuid.uuid4().hex, + 'direction': 'ingress', + 'ethertype': 'IPv4', + 'id': 'security-group-rule-id-' + uuid.uuid4().hex, + 'name': 'security-group-rule-name-' + uuid.uuid4().hex, + 'port_range_max': None, + 'port_range_min': None, + 'protocol': None, + 'remote_group_id': 'remote-security-group-id-' + uuid.uuid4().hex, + 'remote_ip_prefix': None, + 'security_group_id': 'security-group-id-' + uuid.uuid4().hex, + 'tenant_id': 'project-id-' + uuid.uuid4().hex, + } + + # Overwrite default attributes. + security_group_rule_attrs.update(attrs) - :param List security groups: - A list of FakeResource objects faking security groups + # Set default methods. + security_group_rule_methods = {} + + # Overwrite default methods. + security_group_rule_methods.update(methods) + + security_group_rule = fakes.FakeResource( + info=copy.deepcopy(security_group_rule_attrs), + methods=copy.deepcopy(security_group_rule_methods), + loaded=True) + return security_group_rule + + @staticmethod + def create_security_group_rules(attrs={}, methods={}, count=2): + """Create multiple fake security group rules. + + :param Dictionary attrs: + A dictionary with all attributes + :param Dictionary methods: + A dictionary with all methods :param int count: - The number of security groups to fake + The number of security group rules to fake :return: - An iterable Mock object with side_effect set to a list of faked - security groups + A list of FakeResource objects faking the security group rules """ - if security_groups is None: - security_groups = FakeRouter.create_security_groups(count) - return mock.MagicMock(side_effect=security_groups) + security_group_rules = [] + for i in range(0, count): + security_group_rules.append( + FakeSecurityGroupRule.create_one_security_group_rule( + attrs, methods)) + + return security_group_rules class FakeSubnet(object): diff --git a/openstackclient/tests/network/v2/test_network.py b/openstackclient/tests/network/v2/test_network.py index f7721951..26a9da40 100644 --- a/openstackclient/tests/network/v2/test_network.py +++ b/openstackclient/tests/network/v2/test_network.py @@ -579,7 +579,7 @@ class TestNetworkCompute(compute_fakes.TestComputev2): class TestDeleteNetworkCompute(TestNetworkCompute): # The network to delete. - _network = network_fakes.FakeNetwork.create_one_network() + _network = compute_fakes.FakeNetwork.create_one_network() def setUp(self): super(TestDeleteNetworkCompute, self).setUp() @@ -596,10 +596,10 @@ class TestDeleteNetworkCompute(TestNetworkCompute): def test_network_delete(self): arglist = [ - self._network.name, + self._network.label, ] verifylist = [ - ('network', [self._network.name]), + ('network', [self._network.label]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) @@ -607,3 +607,159 @@ class TestDeleteNetworkCompute(TestNetworkCompute): self.compute.networks.delete.assert_called_with(self._network.id) self.assertIsNone(result) + + +class TestListNetworkCompute(TestNetworkCompute): + + # The networks going to be listed up. + _networks = compute_fakes.FakeNetwork.create_networks(count=3) + + columns = ( + 'ID', + 'Name', + 'Subnet', + ) + + data = [] + for net in _networks: + data.append(( + net.id, + net.label, + net.cidr, + )) + + def setUp(self): + super(TestListNetworkCompute, self).setUp() + + self.app.client_manager.network_endpoint_enabled = False + + self.compute.networks.list.return_value = self._networks + + # Get the command object to test + self.cmd = network.ListNetwork(self.app, None) + + def test_network_list_no_options(self): + arglist = [] + verifylist = [ + ('external', False), + ('long', False), + ] + 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) + + self.compute.networks.list.assert_called_with() + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + +class TestShowNetworkCompute(TestNetworkCompute): + + # The network to show. + _network = compute_fakes.FakeNetwork.create_one_network() + + columns = ( + 'bridge', + 'bridge_interface', + 'broadcast', + 'cidr', + 'cidr_v6', + 'created_at', + 'deleted', + 'deleted_at', + 'dhcp_server', + 'dhcp_start', + 'dns1', + 'dns2', + 'enable_dhcp', + 'gateway', + 'gateway_v6', + 'host', + 'id', + 'injected', + 'label', + 'mtu', + 'multi_host', + 'netmask', + 'netmask_v6', + 'priority', + 'project_id', + 'rxtx_base', + 'share_address', + 'updated_at', + 'vlan', + 'vpn_private_address', + 'vpn_public_address', + 'vpn_public_port', + ) + + data = ( + _network.bridge, + _network.bridge_interface, + _network.broadcast, + _network.cidr, + _network.cidr_v6, + _network.created_at, + _network.deleted, + _network.deleted_at, + _network.dhcp_server, + _network.dhcp_start, + _network.dns1, + _network.dns2, + _network.enable_dhcp, + _network.gateway, + _network.gateway_v6, + _network.host, + _network.id, + _network.injected, + _network.label, + _network.mtu, + _network.multi_host, + _network.netmask, + _network.netmask_v6, + _network.priority, + _network.project_id, + _network.rxtx_base, + _network.share_address, + _network.updated_at, + _network.vlan, + _network.vpn_private_address, + _network.vpn_public_address, + _network.vpn_public_port, + ) + + def setUp(self): + super(TestShowNetworkCompute, self).setUp() + + self.app.client_manager.network_endpoint_enabled = False + + # Return value of utils.find_resource() + self.compute.networks.get.return_value = self._network + + # Get the command object to test + self.cmd = network.ShowNetwork(self.app, None) + + def test_show_no_options(self): + arglist = [] + verifylist = [] + + # Missing required args should bail here + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_show_all_options(self): + arglist = [ + self._network.label, + ] + verifylist = [ + ('network', self._network.label), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.assertEqual(self.columns, tuple(columns)) + self.assertEqual(self.data, data) diff --git a/openstackclient/tests/network/v2/test_security_group_rule.py b/openstackclient/tests/network/v2/test_security_group_rule.py new file mode 100644 index 00000000..e0706632 --- /dev/null +++ b/openstackclient/tests/network/v2/test_security_group_rule.py @@ -0,0 +1,100 @@ +# 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 + +from openstackclient.network.v2 import security_group_rule +from openstackclient.tests.compute.v2 import fakes as compute_fakes +from openstackclient.tests.network.v2 import fakes as network_fakes + + +class TestSecurityGroupRuleNetwork(network_fakes.TestNetworkV2): + + def setUp(self): + super(TestSecurityGroupRuleNetwork, self).setUp() + + # Get a shortcut to the network client + self.network = self.app.client_manager.network + + +class TestSecurityGroupRuleCompute(compute_fakes.TestComputev2): + + def setUp(self): + super(TestSecurityGroupRuleCompute, self).setUp() + + # Get a shortcut to the network client + self.compute = self.app.client_manager.compute + + +class TestDeleteSecurityGroupRuleNetwork(TestSecurityGroupRuleNetwork): + + # The security group rule to be deleted. + _security_group_rule = \ + network_fakes.FakeSecurityGroupRule.create_one_security_group_rule() + + def setUp(self): + super(TestDeleteSecurityGroupRuleNetwork, self).setUp() + + self.network.delete_security_group_rule = mock.Mock(return_value=None) + + self.network.find_security_group_rule = mock.Mock( + return_value=self._security_group_rule) + + # Get the command object to test + self.cmd = security_group_rule.DeleteSecurityGroupRule( + self.app, self.namespace) + + def test_security_group_rule_delete(self): + arglist = [ + self._security_group_rule.id, + ] + verifylist = [ + ('rule', self._security_group_rule.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.network.delete_security_group_rule.assert_called_with( + self._security_group_rule) + self.assertEqual(None, result) + + +class TestDeleteSecurityGroupRuleCompute(TestSecurityGroupRuleCompute): + + # The security group rule to be deleted. + _security_group_rule = \ + compute_fakes.FakeSecurityGroupRule.create_one_security_group_rule() + + def setUp(self): + super(TestDeleteSecurityGroupRuleCompute, self).setUp() + + self.app.client_manager.network_endpoint_enabled = False + + # Get the command object to test + self.cmd = security_group_rule.DeleteSecurityGroupRule(self.app, None) + + def test_security_group_rule_delete(self): + arglist = [ + self._security_group_rule.id, + ] + verifylist = [ + ('rule', self._security_group_rule.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.compute.security_group_rules.delete.assert_called_with( + self._security_group_rule.id) + self.assertEqual(None, result) |