diff options
Diffstat (limited to 'openstackclient/tests/network')
| -rw-r--r-- | openstackclient/tests/network/test_common.py | 103 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/fakes.py | 140 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/test_network.py | 41 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/test_port.py | 87 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/test_router.py | 28 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/test_security_group.py | 99 |
6 files changed, 483 insertions, 15 deletions
diff --git a/openstackclient/tests/network/test_common.py b/openstackclient/tests/network/test_common.py new file mode 100644 index 00000000..a3396b9d --- /dev/null +++ b/openstackclient/tests/network/test_common.py @@ -0,0 +1,103 @@ +# 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 argparse +import mock + +from openstackclient.network import common +from openstackclient.tests import utils + + +class FakeNetworkAndComputeCommand(common.NetworkAndComputeCommand): + def update_parser_common(self, parser): + parser.add_argument( + 'common', + metavar='<common>', + help='Common argument', + ) + return parser + + def update_parser_network(self, parser): + parser.add_argument( + 'network', + metavar='<network>', + help='Network argument', + ) + return parser + + def update_parser_compute(self, parser): + parser.add_argument( + 'compute', + metavar='<compute>', + help='Compute argument', + ) + return parser + + def take_action_network(self, client, parsed_args): + client.network_action(parsed_args) + return 'take_action_network' + + def take_action_compute(self, client, parsed_args): + client.compute_action(parsed_args) + return 'take_action_compute' + + +class TestNetworkAndComputeCommand(utils.TestCommand): + def setUp(self): + super(TestNetworkAndComputeCommand, self).setUp() + + self.namespace = argparse.Namespace() + + # Create network client mocks. + self.app.client_manager.network = mock.Mock() + self.network = self.app.client_manager.network + self.network.network_action = mock.Mock(return_value=None) + + # Create compute client mocks. + self.app.client_manager.compute = mock.Mock() + self.compute = self.app.client_manager.compute + self.compute.compute_action = mock.Mock(return_value=None) + + # Get the command object to test + self.cmd = FakeNetworkAndComputeCommand(self.app, self.namespace) + + def test_take_action_network(self): + arglist = [ + 'common', + 'network' + ] + verifylist = [ + ('common', 'common'), + ('network', 'network') + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + self.network.network_action.assert_called_with(parsed_args) + self.assertEqual('take_action_network', result) + + def test_take_action_compute(self): + arglist = [ + 'common', + 'compute' + ] + verifylist = [ + ('common', 'common'), + ('compute', 'compute') + ] + + self.app.client_manager.network_endpoint_enabled = False + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + self.compute.compute_action.assert_called_with(parsed_args) + self.assertEqual('take_action_compute', result) diff --git a/openstackclient/tests/network/v2/fakes.py b/openstackclient/tests/network/v2/fakes.py index 6085bfbd..f07d9d3e 100644 --- a/openstackclient/tests/network/v2/fakes.py +++ b/openstackclient/tests/network/v2/fakes.py @@ -26,6 +26,18 @@ extension_updated = '2013-07-09T12:00:0-00:00' extension_alias = 'Dystopian' extension_links = '[{"href":''"https://github.com/os/network", "type"}]' +QUOTA = { + "subnet": 10, + "network": 10, + "floatingip": 50, + "subnetpool": -1, + "security_group_rule": 100, + "security_group": 10, + "router": 10, + "rbac_policy": -1, + "port": 50, +} + def create_extension(): extension = mock.Mock() @@ -84,7 +96,8 @@ class FakeNetwork(object): 'subnets': ['a', 'b'], 'provider_network_type': 'vlan', 'router_external': True, - 'is_dirty': True, + 'availability_zones': [], + 'availability_zone_hints': [], } # Overwrite default attributes. @@ -93,7 +106,8 @@ class FakeNetwork(object): # Set default methods. network_methods = { 'keys': ['id', 'name', 'admin_state_up', 'router_external', - 'status', 'subnets', 'tenant_id'], + 'status', 'subnets', 'tenant_id', 'availability_zones', + 'availability_zone_hints'], } # Overwrite default methods. @@ -157,15 +171,31 @@ class FakePort(object): :param Dictionary methods: A dictionary with all methods :return: - A FakeResource object, with id, name, admin_state_up, - status, tenant_id + A FakeResource object, with id, name, etc. """ + # Set default attributes. port_attrs = { + 'admin_state_up': True, + 'allowed_address_pairs': [{}], + 'binding:host_id': 'binding-host-id-' + uuid.uuid4().hex, + 'binding:profile': {}, + 'binding:vif_details': {}, + 'binding:vif_type': 'ovs', + 'binding:vnic_type': 'normal', + 'device_id': 'device-id-' + uuid.uuid4().hex, + 'device_owner': 'compute:nova', + 'dns_assignment': [{}], + 'dns_name': 'dns-name-' + uuid.uuid4().hex, + 'extra_dhcp_opts': [{}], + 'fixed_ips': [{}], 'id': 'port-id-' + uuid.uuid4().hex, + 'mac_address': 'fa:16:3e:a9:4e:72', 'name': 'port-name-' + uuid.uuid4().hex, + 'network_id': 'network-id-' + uuid.uuid4().hex, + 'port_security_enabled': True, + 'security_groups': [], 'status': 'ACTIVE', - 'admin_state_up': True, 'tenant_id': 'project-id-' + uuid.uuid4().hex, } @@ -173,7 +203,16 @@ class FakePort(object): port_attrs.update(attrs) # Set default methods. - port_methods = {} + port_methods = { + 'keys': ['admin_state_up', 'allowed_address_pairs', + 'binding:host_id', 'binding:profile', + 'binding:vif_details', 'binding:vif_type', + 'binding:vnic_type', 'device_id', 'device_owner', + 'dns_assignment', 'dns_name', 'extra_dhcp_opts', + 'fixed_ips', 'id', 'mac_address', 'name', + 'network_id', 'port_security_enabled', + 'security_groups', 'status', 'tenant_id'], + } # Overwrite default methods. port_methods.update(methods) @@ -181,6 +220,15 @@ class FakePort(object): port = fakes.FakeResource(info=copy.deepcopy(port_attrs), methods=copy.deepcopy(port_methods), loaded=True) + + # Set attributes with special mappings. + port.project_id = port_attrs['tenant_id'] + port.binding_host_id = port_attrs['binding:host_id'] + port.binding_profile = port_attrs['binding:profile'] + port.binding_vif_details = port_attrs['binding:vif_details'] + port.binding_vif_type = port_attrs['binding:vif_type'] + port.binding_vnic_type = port_attrs['binding:vnic_type'] + return port @staticmethod @@ -248,6 +296,8 @@ class FakeRouter(object): 'tenant_id': 'project-id-' + uuid.uuid4().hex, 'routes': [], 'external_gateway_info': {}, + 'availability_zone_hints': [], + 'availability_zones': [], } # Overwrite default attributes. @@ -306,6 +356,84 @@ class FakeRouter(object): return mock.MagicMock(side_effect=routers) +class FakeSecurityGroup(object): + """Fake one or more security groups.""" + + @staticmethod + def create_one_security_group(attrs={}, methods={}): + """Create a fake security group. + + :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_attrs = { + 'id': 'security-group-id-' + uuid.uuid4().hex, + 'name': 'security-group-name-' + uuid.uuid4().hex, + 'description': 'security-group-description-' + uuid.uuid4().hex, + 'tenant_id': 'project-id-' + uuid.uuid4().hex, + 'security_group_rules': [], + } + + # Overwrite default attributes. + security_group_attrs.update(attrs) + + # Set default methods. + security_group_methods = {} + + # Overwrite default methods. + security_group_methods.update(methods) + + security_group = fakes.FakeResource( + info=copy.deepcopy(security_group_attrs), + methods=copy.deepcopy(security_group_methods), + loaded=True) + return security_group + + @staticmethod + def create_security_groups(attrs={}, methods={}, count=2): + """Create multiple fake security groups. + + :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 + :return: + A list of FakeResource objects faking the security groups + """ + security_groups = [] + for i in range(0, count): + security_groups.append( + FakeRouter.create_one_security_group(attrs, methods)) + + return security_groups + + @staticmethod + def get_security_groups(security_groups=None, count=2): + """Get an iterable MagicMock object with a list of faked security groups. + + If security group list is provided, then initialize the Mock object + with the list. Otherwise create one. + + :param List security groups: + A list of FakeResource objects faking security groups + :param int count: + The number of security groups to fake + :return: + An iterable Mock object with side_effect set to a list of faked + security groups + """ + if security_groups is None: + security_groups = FakeRouter.create_security_groups(count) + return mock.MagicMock(side_effect=security_groups) + + class FakeSubnet(object): """Fake one or more subnets.""" diff --git a/openstackclient/tests/network/v2/test_network.py b/openstackclient/tests/network/v2/test_network.py index 12ac802c..f96497a4 100644 --- a/openstackclient/tests/network/v2/test_network.py +++ b/openstackclient/tests/network/v2/test_network.py @@ -37,11 +37,16 @@ class TestCreateNetworkIdentityV3(TestNetwork): # The new network created. _network = network_fakes.FakeNetwork.create_one_network( - attrs={'tenant_id': identity_fakes_v3.project_id} + attrs={ + 'tenant_id': identity_fakes_v3.project_id, + 'availability_zone_hints': ["nova"], + } ) columns = ( 'admin_state_up', + 'availability_zone_hints', + 'availability_zones', 'id', 'name', 'project_id', @@ -52,6 +57,8 @@ class TestCreateNetworkIdentityV3(TestNetwork): data = ( network._format_admin_state(_network.admin_state_up), + utils.format_list(_network.availability_zone_hints), + utils.format_list(_network.availability_zones), _network.id, _network.name, _network.project_id, @@ -129,6 +136,7 @@ class TestCreateNetworkIdentityV3(TestNetwork): "--share", "--project", identity_fakes_v3.project_name, "--project-domain", identity_fakes_v3.domain_name, + "--availability-zone-hint", "nova", self._network.name, ] verifylist = [ @@ -136,6 +144,7 @@ class TestCreateNetworkIdentityV3(TestNetwork): ('shared', True), ('project', identity_fakes_v3.project_name), ('project_domain', identity_fakes_v3.domain_name), + ('availability_zone_hints', ["nova"]), ('name', self._network.name), ] @@ -144,6 +153,7 @@ class TestCreateNetworkIdentityV3(TestNetwork): self.network.create_network.assert_called_with(**{ 'admin_state_up': False, + 'availability_zone_hints': ["nova"], 'name': self._network.name, 'shared': True, 'tenant_id': identity_fakes_v3.project_id, @@ -184,6 +194,8 @@ class TestCreateNetworkIdentityV2(TestNetwork): columns = ( 'admin_state_up', + 'availability_zone_hints', + 'availability_zones', 'id', 'name', 'project_id', @@ -194,6 +206,8 @@ class TestCreateNetworkIdentityV2(TestNetwork): data = ( network._format_admin_state(_network.admin_state_up), + utils.format_list(_network.availability_zone_hints), + utils.format_list(_network.availability_zones), _network.id, _network.name, _network.project_id, @@ -324,6 +338,7 @@ class TestListNetwork(TestNetwork): 'Subnets', 'Network Type', 'Router Type', + 'Availability Zones', ) data = [] @@ -346,6 +361,7 @@ class TestListNetwork(TestNetwork): utils.format_list(net.subnets), net.provider_network_type, network._format_router_external(net.router_external), + utils.format_list(net.availability_zones), )) def setUp(self): @@ -424,8 +440,6 @@ class TestSetNetwork(TestNetwork): self.cmd = network.SetNetwork(self.app, self.namespace) def test_set_this(self): - self._network.is_dirty = True - arglist = [ self._network.name, '--enable', @@ -442,12 +456,15 @@ class TestSetNetwork(TestNetwork): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.network.update_network.assert_called_with(self._network) + attrs = { + 'name': 'noob', + 'admin_state_up': True, + 'shared': True, + } + self.network.update_network.assert_called_with(self._network, **attrs) self.assertIsNone(result) def test_set_that(self): - self._network.is_dirty = True - arglist = [ self._network.name, '--disable', @@ -462,12 +479,14 @@ class TestSetNetwork(TestNetwork): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.network.update_network.assert_called_with(self._network) + attrs = { + 'admin_state_up': False, + 'shared': False, + } + self.network.update_network.assert_called_with(self._network, **attrs) self.assertIsNone(result) def test_set_nothing(self): - self._network.is_dirty = False - arglist = [self._network.name, ] verifylist = [('identifier', self._network.name), ] @@ -483,6 +502,8 @@ class TestShowNetwork(TestNetwork): columns = ( 'admin_state_up', + 'availability_zone_hints', + 'availability_zones', 'id', 'name', 'project_id', @@ -493,6 +514,8 @@ class TestShowNetwork(TestNetwork): data = ( network._format_admin_state(_network.admin_state_up), + utils.format_list(_network.availability_zone_hints), + utils.format_list(_network.availability_zones), _network.id, _network.name, _network.project_id, diff --git a/openstackclient/tests/network/v2/test_port.py b/openstackclient/tests/network/v2/test_port.py index a1ddefa1..bc246bd8 100644 --- a/openstackclient/tests/network/v2/test_port.py +++ b/openstackclient/tests/network/v2/test_port.py @@ -13,8 +13,10 @@ import mock +from openstackclient.common import utils from openstackclient.network.v2 import port from openstackclient.tests.network.v2 import fakes as network_fakes +from openstackclient.tests import utils as tests_utils class TestPort(network_fakes.TestNetworkV2): @@ -51,3 +53,88 @@ class TestDeletePort(TestPort): result = self.cmd.take_action(parsed_args) self.network.delete_port.assert_called_with(self._port) self.assertIsNone(result) + + +class TestShowPort(TestPort): + + # The port to show. + _port = network_fakes.FakePort.create_one_port() + + columns = ( + 'admin_state_up', + 'allowed_address_pairs', + 'binding_host_id', + 'binding_profile', + 'binding_vif_details', + 'binding_vif_type', + 'binding_vnic_type', + 'device_id', + 'device_owner', + 'dns_assignment', + 'dns_name', + 'extra_dhcp_opts', + 'fixed_ips', + 'id', + 'mac_address', + 'name', + 'network_id', + 'port_security_enabled', + 'project_id', + 'security_groups', + 'status', + ) + + data = ( + port._format_admin_state(_port.admin_state_up), + utils.format_list_of_dicts(_port.allowed_address_pairs), + _port.binding_host_id, + utils.format_dict(_port.binding_profile), + utils.format_dict(_port.binding_vif_details), + _port.binding_vif_type, + _port.binding_vnic_type, + _port.device_id, + _port.device_owner, + utils.format_list_of_dicts(_port.dns_assignment), + _port.dns_name, + utils.format_list_of_dicts(_port.extra_dhcp_opts), + utils.format_list_of_dicts(_port.fixed_ips), + _port.id, + _port.mac_address, + _port.name, + _port.network_id, + _port.port_security_enabled, + _port.project_id, + utils.format_list(_port.security_groups), + _port.status, + ) + + def setUp(self): + super(TestShowPort, self).setUp() + + self.network.find_port = mock.Mock(return_value=self._port) + + # Get the command object to test + self.cmd = port.ShowPort(self.app, self.namespace) + + def test_show_no_options(self): + arglist = [] + verifylist = [] + + self.assertRaises(tests_utils.ParserException, + self.check_parser, self.cmd, arglist, verifylist) + + def test_show_all_options(self): + arglist = [ + self._port.name, + ] + verifylist = [ + ('port', self._port.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.network.find_port.assert_called_with(self._port.name, + ignore_missing=False) + self.assertEqual(tuple(self.columns), columns) + self.assertEqual(self.data, data) diff --git a/openstackclient/tests/network/v2/test_router.py b/openstackclient/tests/network/v2/test_router.py index fba6e192..98e9f17a 100644 --- a/openstackclient/tests/network/v2/test_router.py +++ b/openstackclient/tests/network/v2/test_router.py @@ -14,6 +14,7 @@ import mock from openstackclient.common import exceptions +from openstackclient.common import utils as osc_utils from openstackclient.network.v2 import router from openstackclient.tests.network.v2 import fakes as network_fakes from openstackclient.tests import utils as tests_utils @@ -88,6 +89,31 @@ class TestCreateRouter(TestRouter): self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) + def test_create_with_AZ_hints(self): + arglist = [ + self.new_router.name, + '--availability-zone-hint', 'fake-az', + '--availability-zone-hint', 'fake-az2', + ] + verifylist = [ + ('name', self.new_router.name), + ('availability_zone_hints', ['fake-az', 'fake-az2']), + ('admin_state_up', True), + ('distributed', False), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = (self.cmd.take_action(parsed_args)) + self.network.create_router.assert_called_with(**{ + 'admin_state_up': True, + 'name': self.new_router.name, + 'distributed': False, + 'availability_zone_hints': ['fake-az', 'fake-az2'], + }) + + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + class TestDeleteRouter(TestRouter): @@ -135,6 +161,7 @@ class TestListRouter(TestRouter): columns_long = columns + ( 'Routes', 'External gateway info', + 'Availability zones' ) data = [] @@ -155,6 +182,7 @@ class TestListRouter(TestRouter): data[i] + ( r.routes, router._format_external_gateway_info(r.external_gateway_info), + osc_utils.format_list(r.availability_zones), ) ) diff --git a/openstackclient/tests/network/v2/test_security_group.py b/openstackclient/tests/network/v2/test_security_group.py new file mode 100644 index 00000000..98388ec7 --- /dev/null +++ b/openstackclient/tests/network/v2/test_security_group.py @@ -0,0 +1,99 @@ +# 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 +from openstackclient.tests.network.v2 import fakes as network_fakes + + +class TestSecurityGroup(network_fakes.TestNetworkV2): + + def setUp(self): + super(TestSecurityGroup, self).setUp() + + # Get a shortcut to the network client + self.network = self.app.client_manager.network + + # Create compute client mocks. + self.app.client_manager.compute = mock.Mock() + self.compute = self.app.client_manager.compute + self.compute.security_groups = mock.Mock() + + +class TestDeleteSecurityGroupNetwork(TestSecurityGroup): + + # The security group to be deleted. + _security_group = \ + network_fakes.FakeSecurityGroup.create_one_security_group() + + def setUp(self): + super(TestDeleteSecurityGroupNetwork, self).setUp() + + self.network.delete_security_group = mock.Mock(return_value=None) + + self.network.find_security_group = mock.Mock( + return_value=self._security_group) + + # Get the command object to test + self.cmd = security_group.DeleteSecurityGroup(self.app, self.namespace) + + def test_security_group_delete(self): + arglist = [ + self._security_group.name, + ] + verifylist = [ + ('group', self._security_group.name), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + self.network.delete_security_group.assert_called_with( + self._security_group) + self.assertEqual(None, result) + + +class TestDeleteSecurityGroupCompute(TestSecurityGroup): + + # The security group to be deleted. + _security_group = \ + network_fakes.FakeSecurityGroup.create_one_security_group() + + def setUp(self): + super(TestDeleteSecurityGroupCompute, self).setUp() + + self.app.client_manager.network_endpoint_enabled = False + + self.compute.security_groups.delete = mock.Mock(return_value=None) + + self.compute.security_groups.get = mock.Mock( + return_value=self._security_group) + + # Get the command object to test + self.cmd = security_group.DeleteSecurityGroup(self.app, self.namespace) + + def test_security_group_delete(self): + arglist = [ + self._security_group.name, + ] + verifylist = [ + ('group', self._security_group.name), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + self.compute.security_groups.delete.assert_called_with( + self._security_group.id) + self.assertEqual(None, result) |