diff options
Diffstat (limited to 'openstackclient/tests')
17 files changed, 963 insertions, 320 deletions
diff --git a/openstackclient/tests/common/test_command.py b/openstackclient/tests/common/test_command.py index a8bcf6a8..658bc895 100644 --- a/openstackclient/tests/common/test_command.py +++ b/openstackclient/tests/common/test_command.py @@ -14,9 +14,9 @@ import mock -from osc_lib.command import command from osc_lib import exceptions +from openstackclient.common import command from openstackclient.tests import fakes as test_fakes from openstackclient.tests import utils as test_utils diff --git a/openstackclient/tests/common/test_timing.py b/openstackclient/tests/common/test_timing.py deleted file mode 100644 index e33bb7ae..00000000 --- a/openstackclient/tests/common/test_timing.py +++ /dev/null @@ -1,94 +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. -# - -"""Test Timing pseudo-command""" - -import datetime - -from openstackclient.common import timing -from openstackclient.tests import fakes -from openstackclient.tests import utils - - -timing_url = 'GET http://localhost:5000' -timing_elapsed = 0.872809 - - -class FakeGenericClient(object): - - def __init__(self, **kwargs): - self.auth_token = kwargs['token'] - self.management_url = kwargs['endpoint'] - - -class TestTiming(utils.TestCommand): - - columns = ( - 'URL', - 'Seconds', - ) - - def setUp(self): - super(TestTiming, self).setUp() - - self.app.timing_data = [] - - self.app.client_manager.compute = FakeGenericClient( - endpoint=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN, - ) - - self.app.client_manager.volume = FakeGenericClient( - endpoint=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN, - ) - - # Get the command object to test - self.cmd = timing.Timing(self.app, None) - - def test_timing_list_no_data(self): - arglist = [] - verifylist = [] - 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.assertEqual(self.columns, columns) - datalist = [ - ('Total', 0.0,) - ] - self.assertEqual(datalist, data) - - def test_timing_list(self): - self.app.timing_data = [( - timing_url, - datetime.timedelta(microseconds=timing_elapsed * 1000000), - )] - - arglist = [] - verifylist = [] - 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.assertEqual(self.columns, columns) - datalist = [ - (timing_url, timing_elapsed), - ('Total', timing_elapsed), - ] - self.assertEqual(datalist, data) diff --git a/openstackclient/tests/compute/v2/fakes.py b/openstackclient/tests/compute/v2/fakes.py index 9682eec4..60abb8ef 100644 --- a/openstackclient/tests/compute/v2/fakes.py +++ b/openstackclient/tests/compute/v2/fakes.py @@ -452,6 +452,25 @@ class FakeSecurityGroup(object): 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 groups 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 = FakeSecurityGroup.create_security_groups(count) + return mock.MagicMock(side_effect=security_groups) + class FakeSecurityGroupRule(object): """Fake one or more security group rules.""" diff --git a/openstackclient/tests/compute/v2/test_flavor.py b/openstackclient/tests/compute/v2/test_flavor.py index 27b53bbf..4365a540 100644 --- a/openstackclient/tests/compute/v2/test_flavor.py +++ b/openstackclient/tests/compute/v2/test_flavor.py @@ -469,6 +469,7 @@ class TestFlavorSet(TestFlavor): result = self.cmd.take_action(parsed_args) self.flavors_mock.find.assert_called_with(name=parsed_args.flavor, is_public=None) + self.flavor.set_keys.assert_called_with({'FOO': '"B A R"'}) self.assertIsNone(result) def test_flavor_set_project(self): @@ -483,12 +484,15 @@ class TestFlavorSet(TestFlavor): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.assertIsNone(result) + self.flavors_mock.find.assert_called_with(name=parsed_args.flavor, + is_public=None) self.flavor_access_mock.add_tenant_access.assert_called_with( self.flavor.id, identity_fakes.project_id, ) + self.flavor.set_keys.assert_not_called() + self.assertIsNone(result) def test_flavor_set_no_project(self): arglist = [ @@ -496,7 +500,7 @@ class TestFlavorSet(TestFlavor): self.flavor.id, ] verifylist = [ - ('project', ''), + ('project', None), ('flavor', self.flavor.id), ] self.assertRaises(tests_utils.ParserException, self.check_parser, @@ -509,12 +513,8 @@ class TestFlavorSet(TestFlavor): verifylist = [ ('project', identity_fakes.project_id), ] - - self.assertRaises(tests_utils.ParserException, - self.check_parser, - self.cmd, - arglist, - verifylist) + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) def test_flavor_set_with_unexist_flavor(self): self.flavors_mock.get.side_effect = exceptions.NotFound(None) @@ -541,7 +541,6 @@ class TestFlavorSet(TestFlavor): verifylist = [ ('flavor', self.flavor.id), ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) self.assertRaises(exceptions.CommandError, self.cmd.take_action, parsed_args) @@ -672,6 +671,18 @@ class TestFlavorUnset(TestFlavor): self.flavor.unset_keys.assert_not_called() self.assertIsNone(result) + def test_flavor_unset_no_project(self): + arglist = [ + '--project', + self.flavor.id, + ] + verifylist = [ + ('project', None), + ('flavor', self.flavor.id), + ] + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + def test_flavor_unset_no_flavor(self): arglist = [ '--project', identity_fakes.project_id, diff --git a/openstackclient/tests/compute/v2/test_server.py b/openstackclient/tests/compute/v2/test_server.py index e10f43a1..0f155601 100644 --- a/openstackclient/tests/compute/v2/test_server.py +++ b/openstackclient/tests/compute/v2/test_server.py @@ -511,150 +511,6 @@ class TestServerDumpCreate(TestServer): self.run_method_with_servers('trigger_crash_dump', 3) -class TestServerImageCreate(TestServer): - - columns = ( - 'id', - 'name', - 'owner', - 'protected', - 'tags', - 'visibility', - ) - - def datalist(self): - datalist = ( - self.image.id, - self.image.name, - self.image.owner, - self.image.protected, - self.image.tags, - self.image.visibility, - ) - return datalist - - def setUp(self): - super(TestServerImageCreate, self).setUp() - - self.server = compute_fakes.FakeServer.create_one_server() - - # This is the return value for utils.find_resource() - self.servers_mock.get.return_value = self.server - - self.image = image_fakes.FakeImage.create_one_image() - self.images_mock.get.return_value = self.image - self.servers_mock.create_image.return_value = self.image.id - - # Get the command object to test - self.cmd = server.CreateServerImage(self.app, None) - - def test_server_image_create_no_options(self): - arglist = [ - self.server.id, - ] - verifylist = [ - ('server', self.server.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # ServerManager.create_image(server, image_name, metadata=) - self.servers_mock.create_image.assert_called_with( - self.servers_mock.get.return_value, - self.server.name, - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist(), data) - - def test_server_image_create_name(self): - arglist = [ - '--name', 'img-nam', - self.server.id, - ] - verifylist = [ - ('name', 'img-nam'), - ('server', self.server.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # ServerManager.create_image(server, image_name, metadata=) - self.servers_mock.create_image.assert_called_with( - self.servers_mock.get.return_value, - 'img-nam', - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist(), data) - - @mock.patch.object(common_utils, 'wait_for_status', return_value=False) - def test_server_create_image_with_wait_fails(self, mock_wait_for_status): - arglist = [ - '--wait', - self.server.id, - ] - verifylist = [ - ('wait', True), - ('server', self.server.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - self.assertRaises(SystemExit, self.cmd.take_action, parsed_args) - - mock_wait_for_status.assert_called_once_with( - self.images_mock.get, - self.image.id, - callback=server._show_progress - ) - - # ServerManager.create_image(server, image_name, metadata=) - self.servers_mock.create_image.assert_called_with( - self.servers_mock.get.return_value, - self.server.name, - ) - - @mock.patch.object(common_utils, 'wait_for_status', return_value=True) - def test_server_create_image_with_wait_ok(self, mock_wait_for_status): - arglist = [ - '--wait', - self.server.id, - ] - verifylist = [ - ('wait', True), - ('server', self.server.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns a two-part tuple with a tuple of column names and a tuple of - # data to be shown. - columns, data = self.cmd.take_action(parsed_args) - - # ServerManager.create_image(server, image_name, metadata=) - self.servers_mock.create_image.assert_called_with( - self.servers_mock.get.return_value, - self.server.name, - ) - - mock_wait_for_status.assert_called_once_with( - self.images_mock.get, - self.image.id, - callback=server._show_progress - ) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.datalist(), data) - - class TestServerList(TestServer): # Columns to be listed up. diff --git a/openstackclient/tests/compute/v2/test_server_image.py b/openstackclient/tests/compute/v2/test_server_image.py new file mode 100644 index 00000000..660e9817 --- /dev/null +++ b/openstackclient/tests/compute/v2/test_server_image.py @@ -0,0 +1,227 @@ +# 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.common import exceptions +from openstackclient.common import utils as common_utils +from openstackclient.compute.v2 import server_image +from openstackclient.tests.compute.v2 import fakes as compute_fakes +from openstackclient.tests.image.v2 import fakes as image_fakes + + +class TestServerImage(compute_fakes.TestComputev2): + + def setUp(self): + super(TestServerImage, self).setUp() + + # Get a shortcut to the compute client ServerManager Mock + self.servers_mock = self.app.client_manager.compute.servers + self.servers_mock.reset_mock() + + # Get a shortcut to the image client ImageManager Mock + self.images_mock = self.app.client_manager.image.images + self.images_mock.reset_mock() + + # Set object attributes to be tested. Could be overwriten in subclass. + self.attrs = {} + + # Set object methods to be tested. Could be overwriten in subclass. + self.methods = {} + + def setup_servers_mock(self, count): + servers = compute_fakes.FakeServer.create_servers( + attrs=self.attrs, + methods=self.methods, + count=count, + ) + + # This is the return value for utils.find_resource() + self.servers_mock.get = compute_fakes.FakeServer.get_servers( + servers, + 0, + ) + return servers + + +class TestServerImageCreate(TestServerImage): + + def image_columns(self, image): + columnlist = tuple(sorted(image.keys())) + return columnlist + + def image_data(self, image): + datalist = ( + image['id'], + image['name'], + image['owner'], + image['protected'], + 'active', + common_utils.format_list(image.get('tags')), + image['visibility'], + ) + return datalist + + def setUp(self): + super(TestServerImageCreate, self).setUp() + + # Get the command object to test + self.cmd = server_image.CreateServerImage(self.app, None) + + self.methods = { + 'create_image': None, + } + + def setup_images_mock(self, count, servers=None): + if servers: + images = image_fakes.FakeImage.create_images( + attrs={ + 'name': servers[0].name, + 'status': 'active', + }, + count=count, + ) + else: + images = image_fakes.FakeImage.create_images( + attrs={ + 'status': 'active', + }, + count=count, + ) + + self.images_mock.get = mock.MagicMock(side_effect=images) + self.servers_mock.create_image = mock.MagicMock( + return_value=images[0].id, + ) + return images + + def test_server_image_create_defaults(self): + servers = self.setup_servers_mock(count=1) + images = self.setup_images_mock(count=1, servers=servers) + + arglist = [ + servers[0].id, + ] + verifylist = [ + ('server', servers[0].id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # In base command class ShowOne in cliff, abstract method take_action() + # returns a two-part tuple with a tuple of column names and a tuple of + # data to be shown. + columns, data = self.cmd.take_action(parsed_args) + + # ServerManager.create_image(server, image_name, metadata=) + self.servers_mock.create_image.assert_called_with( + servers[0].id, + servers[0].name, + ) + + self.assertEqual(self.image_columns(images[0]), columns) + self.assertEqual(self.image_data(images[0]), data) + + def test_server_image_create_options(self): + servers = self.setup_servers_mock(count=1) + images = self.setup_images_mock(count=1, servers=servers) + + arglist = [ + '--name', 'img-nam', + servers[0].id, + ] + verifylist = [ + ('name', 'img-nam'), + ('server', servers[0].id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # In base command class ShowOne in cliff, abstract method take_action() + # returns a two-part tuple with a tuple of column names and a tuple of + # data to be shown. + columns, data = self.cmd.take_action(parsed_args) + + # ServerManager.create_image(server, image_name, metadata=) + self.servers_mock.create_image.assert_called_with( + servers[0].id, + 'img-nam', + ) + + self.assertEqual(self.image_columns(images[0]), columns) + self.assertEqual(self.image_data(images[0]), data) + + @mock.patch.object(common_utils, 'wait_for_status', return_value=False) + def test_server_create_image_wait_fail(self, mock_wait_for_status): + servers = self.setup_servers_mock(count=1) + images = self.setup_images_mock(count=1, servers=servers) + + arglist = [ + '--wait', + servers[0].id, + ] + verifylist = [ + ('wait', True), + ('server', servers[0].id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + self.assertRaises( + exceptions.CommandError, + self.cmd.take_action, + parsed_args, + ) + + # ServerManager.create_image(server, image_name, metadata=) + self.servers_mock.create_image.assert_called_with( + servers[0].id, + servers[0].name, + ) + + mock_wait_for_status.assert_called_once_with( + self.images_mock.get, + images[0].id, + callback=mock.ANY + ) + + @mock.patch.object(common_utils, 'wait_for_status', return_value=True) + def test_server_create_image_wait_ok(self, mock_wait_for_status): + servers = self.setup_servers_mock(count=1) + images = self.setup_images_mock(count=1, servers=servers) + + arglist = [ + '--wait', + servers[0].id, + ] + verifylist = [ + ('wait', True), + ('server', servers[0].id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # In base command class ShowOne in cliff, abstract method take_action() + # returns a two-part tuple with a tuple of column names and a tuple of + # data to be shown. + columns, data = self.cmd.take_action(parsed_args) + + # ServerManager.create_image(server, image_name, metadata=) + self.servers_mock.create_image.assert_called_with( + servers[0].id, + servers[0].name, + ) + + mock_wait_for_status.assert_called_once_with( + self.images_mock.get, + images[0].id, + callback=mock.ANY + ) + + self.assertEqual(self.image_columns(images[0]), columns) + self.assertEqual(self.image_data(images[0]), data) diff --git a/openstackclient/tests/compute/v2/test_service.py b/openstackclient/tests/compute/v2/test_service.py index 3f741340..b360c9dc 100644 --- a/openstackclient/tests/compute/v2/test_service.py +++ b/openstackclient/tests/compute/v2/test_service.py @@ -13,6 +13,8 @@ # under the License. # +import mock + from osc_lib import exceptions from openstackclient.compute.v2 import service @@ -225,8 +227,12 @@ class TestServiceSet(TestService): ('service', self.service.binary), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) - self.assertRaises(exceptions.CommandError, self.cmd.take_action, - parsed_args) + try: + self.cmd.take_action(parsed_args) + self.fail("CommandError should be raised.") + except exceptions.CommandError as e: + self.assertEqual("Cannot specify option --disable-reason without " + "--disable specified.", str(e)) def test_service_set_enable_with_disable_reason(self): reason = 'earthquake' @@ -243,5 +249,93 @@ class TestServiceSet(TestService): ('service', self.service.binary), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) - self.assertRaises(exceptions.CommandError, self.cmd.take_action, - parsed_args) + try: + self.cmd.take_action(parsed_args) + self.fail("CommandError should be raised.") + except exceptions.CommandError as e: + self.assertEqual("Cannot specify option --disable-reason without " + "--disable specified.", str(e)) + + def test_service_set_state_up(self): + arglist = [ + '--up', + self.service.host, + self.service.binary, + ] + verifylist = [ + ('up', True), + ('host', self.service.host), + ('service', self.service.binary), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + self.service_mock.force_down.assert_called_once_with( + self.service.host, self.service.binary, force_down=False) + self.assertNotCalled(self.service_mock.enable) + self.assertNotCalled(self.service_mock.disable) + self.assertIsNone(result) + + def test_service_set_state_down(self): + arglist = [ + '--down', + self.service.host, + self.service.binary, + ] + verifylist = [ + ('down', True), + ('host', self.service.host), + ('service', self.service.binary), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + self.service_mock.force_down.assert_called_once_with( + self.service.host, self.service.binary, force_down=True) + self.assertNotCalled(self.service_mock.enable) + self.assertNotCalled(self.service_mock.disable) + self.assertIsNone(result) + + def test_service_set_enable_and_state_down(self): + arglist = [ + '--enable', + '--down', + self.service.host, + self.service.binary, + ] + verifylist = [ + ('enable', True), + ('down', True), + ('host', self.service.host), + ('service', self.service.binary), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + self.service_mock.enable.assert_called_once_with( + self.service.host, self.service.binary) + self.service_mock.force_down.assert_called_once_with( + self.service.host, self.service.binary, force_down=True) + self.assertIsNone(result) + + def test_service_set_enable_and_state_down_with_exception(self): + arglist = [ + '--enable', + '--down', + self.service.host, + self.service.binary, + ] + verifylist = [ + ('enable', True), + ('down', True), + ('host', self.service.host), + ('service', self.service.binary), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + with mock.patch.object(self.cmd.log, 'error') as mock_log: + with mock.patch.object(self.service_mock, 'enable', + side_effect=Exception()): + self.assertRaises(exceptions.CommandError, + self.cmd.take_action, parsed_args) + mock_log.assert_called_once_with( + "Failed to set service status to %s", "enabled") + self.service_mock.force_down.assert_called_once_with( + self.service.host, self.service.binary, force_down=True) diff --git a/openstackclient/tests/image/v2/test_image.py b/openstackclient/tests/image/v2/test_image.py index 3dbf504a..592def21 100644 --- a/openstackclient/tests/image/v2/test_image.py +++ b/openstackclient/tests/image/v2/test_image.py @@ -474,6 +474,37 @@ class TestImageDelete(TestImage): self.images_mock.delete.assert_has_calls(calls) self.assertIsNone(result) + def test_image_delete_multi_images_exception(self): + + images = image_fakes.FakeImage.create_images(count=2) + arglist = [ + images[0].id, + images[1].id, + 'x-y-x', + ] + verifylist = [ + ('images', arglist) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # Fake exception in utils.find_resource() + # In image v2, we use utils.find_resource() to find a network. + # It calls get() several times, but find() only one time. So we + # choose to fake get() always raise exception, then pass through. + # And fake find() to find the real network or not. + ret_find = [ + images[0], + images[1], + exceptions.NotFound('404'), + ] + + self.images_mock.get = Exception() + self.images_mock.find.side_effect = ret_find + self.assertRaises(exceptions.CommandError, self.cmd.take_action, + parsed_args) + calls = [mock.call(i.id) for i in images] + self.images_mock.delete.assert_has_calls(calls) + class TestImageList(TestImage): diff --git a/openstackclient/tests/network/v2/fakes.py b/openstackclient/tests/network/v2/fakes.py index 9efbe8c6..a23efc2d 100644 --- a/openstackclient/tests/network/v2/fakes.py +++ b/openstackclient/tests/network/v2/fakes.py @@ -611,6 +611,25 @@ class FakeSecurityGroup(object): 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 groups 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 = FakeSecurityGroup.create_security_groups(count) + return mock.MagicMock(side_effect=security_groups) + class FakeSecurityGroupRule(object): """Fake one or more security group rules.""" @@ -670,6 +689,26 @@ class FakeSecurityGroupRule(object): return security_group_rules + @staticmethod + def get_security_group_rules(security_group_rules=None, count=2): + """Get an iterable MagicMock object with a list of faked security group rules. + + If security group rules list is provided, then initialize the Mock + object with the list. Otherwise create one. + + :param List security group rules: + A list of FakeResource objects faking security group rules + :param int count: + The number of security group rules to fake + :return: + An iterable Mock object with side_effect set to a list of faked + security group rules + """ + if security_group_rules is None: + security_group_rules = ( + FakeSecurityGroupRule.create_security_group_rules(count)) + return mock.MagicMock(side_effect=security_group_rules) + class FakeSubnet(object): """Fake one or more subnets.""" @@ -732,6 +771,25 @@ class FakeSubnet(object): return subnets + @staticmethod + def get_subnets(subnets=None, count=2): + """Get an iterable MagicMock object with a list of faked subnets. + + If subnets list is provided, then initialize the Mock object + with the list. Otherwise create one. + + :param List subnets: + A list of FakeResource objects faking subnets + :param int count: + The number of subnets to fake + :return: + An iterable Mock object with side_effect set to a list of faked + subnets + """ + if subnets is None: + subnets = FakeSubnet.create_subnets(count) + return mock.MagicMock(side_effect=subnets) + class FakeFloatingIP(object): """Fake one or more floating ip.""" @@ -871,3 +929,22 @@ class FakeSubnetPool(object): ) return subnet_pools + + @staticmethod + def get_subnet_pools(subnet_pools=None, count=2): + """Get an iterable MagicMock object with a list of faked subnet pools. + + If subnet_pools list is provided, then initialize the Mock object + with the list. Otherwise create one. + + :param List subnet pools: + A list of FakeResource objects faking subnet pools + :param int count: + The number of subnet pools to fake + :return: + An iterable Mock object with side_effect set to a list of faked + subnet pools + """ + if subnet_pools is None: + subnet_pools = FakeSubnetPool.create_subnet_pools(count) + return mock.MagicMock(side_effect=subnet_pools) diff --git a/openstackclient/tests/network/v2/test_floating_ip.py b/openstackclient/tests/network/v2/test_floating_ip.py index f9ccfe1c..5cd5279a 100644 --- a/openstackclient/tests/network/v2/test_floating_ip.py +++ b/openstackclient/tests/network/v2/test_floating_ip.py @@ -12,6 +12,9 @@ # import mock +from mock import call + +from osc_lib import exceptions from openstackclient.network.v2 import floating_ip from openstackclient.tests.compute.v2 import fakes as compute_fakes @@ -140,33 +143,84 @@ class TestCreateFloatingIPNetwork(TestFloatingIPNetwork): class TestDeleteFloatingIPNetwork(TestFloatingIPNetwork): - # The floating ip to be deleted. - floating_ip = network_fakes.FakeFloatingIP.create_one_floating_ip() + # The floating ips to be deleted. + floating_ips = network_fakes.FakeFloatingIP.create_floating_ips(count=2) def setUp(self): super(TestDeleteFloatingIPNetwork, self).setUp() self.network.delete_ip = mock.Mock(return_value=None) - self.network.find_ip = mock.Mock(return_value=self.floating_ip) + self.network.find_ip = ( + network_fakes.FakeFloatingIP.get_floating_ips(self.floating_ips)) # Get the command object to test self.cmd = floating_ip.DeleteFloatingIP(self.app, self.namespace) def test_floating_ip_delete(self): arglist = [ - self.floating_ip.id, + self.floating_ips[0].id, ] verifylist = [ - ('floating_ip', self.floating_ip.id), + ('floating_ip', [self.floating_ips[0].id]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.network.find_ip.assert_called_once_with(self.floating_ip.id) - self.network.delete_ip.assert_called_once_with(self.floating_ip) + self.network.find_ip.assert_called_once_with( + self.floating_ips[0].id, ignore_missing=False) + self.network.delete_ip.assert_called_once_with(self.floating_ips[0]) self.assertIsNone(result) + def test_multi_floating_ips_delete(self): + arglist = [] + verifylist = [] + + for f in self.floating_ips: + arglist.append(f.id) + verifylist = [ + ('floating_ip', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for f in self.floating_ips: + calls.append(call(f)) + self.network.delete_ip.assert_has_calls(calls) + self.assertIsNone(result) + + def test_multi_floating_ips_delete_with_exception(self): + arglist = [ + self.floating_ips[0].id, + 'unexist_floating_ip', + ] + verifylist = [ + ('floating_ip', + [self.floating_ips[0].id, 'unexist_floating_ip']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self.floating_ips[0], exceptions.CommandError] + self.network.find_ip = ( + mock.MagicMock(side_effect=find_mock_result) + ) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 floating_ip failed to delete.', str(e)) + + self.network.find_ip.assert_any_call( + self.floating_ips[0].id, ignore_missing=False) + self.network.find_ip.assert_any_call( + 'unexist_floating_ip', ignore_missing=False) + self.network.delete_ip.assert_called_once_with( + self.floating_ips[0] + ) + class TestListFloatingIPNetwork(TestFloatingIPNetwork): @@ -335,8 +389,8 @@ class TestCreateFloatingIPCompute(TestFloatingIPCompute): class TestDeleteFloatingIPCompute(TestFloatingIPCompute): - # The floating ip to be deleted. - floating_ip = compute_fakes.FakeFloatingIP.create_one_floating_ip() + # The floating ips to be deleted. + floating_ips = compute_fakes.FakeFloatingIP.create_floating_ips(count=2) def setUp(self): super(TestDeleteFloatingIPCompute, self).setUp() @@ -346,27 +400,78 @@ class TestDeleteFloatingIPCompute(TestFloatingIPCompute): self.compute.floating_ips.delete.return_value = None # Return value of utils.find_resource() - self.compute.floating_ips.get.return_value = self.floating_ip + self.compute.floating_ips.get = ( + compute_fakes.FakeFloatingIP.get_floating_ips(self.floating_ips)) # Get the command object to test self.cmd = floating_ip.DeleteFloatingIP(self.app, None) def test_floating_ip_delete(self): arglist = [ - self.floating_ip.id, + self.floating_ips[0].id, ] verifylist = [ - ('floating_ip', self.floating_ip.id), + ('floating_ip', [self.floating_ips[0].id]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) self.compute.floating_ips.delete.assert_called_once_with( - self.floating_ip.id + self.floating_ips[0].id ) self.assertIsNone(result) + def test_multi_floating_ips_delete(self): + arglist = [] + verifylist = [] + + for f in self.floating_ips: + arglist.append(f.id) + verifylist = [ + ('floating_ip', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for f in self.floating_ips: + calls.append(call(f.id)) + self.compute.floating_ips.delete.assert_has_calls(calls) + self.assertIsNone(result) + + def test_multi_floating_ips_delete_with_exception(self): + arglist = [ + self.floating_ips[0].id, + 'unexist_floating_ip', + ] + verifylist = [ + ('floating_ip', + [self.floating_ips[0].id, 'unexist_floating_ip']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self.floating_ips[0], exceptions.CommandError] + self.compute.floating_ips.get = ( + mock.MagicMock(side_effect=find_mock_result) + ) + self.compute.floating_ips.find.side_effect = exceptions.NotFound(None) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 floating_ip failed to delete.', str(e)) + + self.compute.floating_ips.get.assert_any_call( + self.floating_ips[0].id) + self.compute.floating_ips.get.assert_any_call( + 'unexist_floating_ip') + self.compute.floating_ips.delete.assert_called_once_with( + self.floating_ips[0].id + ) + class TestListFloatingIPCompute(TestFloatingIPCompute): diff --git a/openstackclient/tests/network/v2/test_ip_availability.py b/openstackclient/tests/network/v2/test_ip_availability.py index 39e11cd5..c6ec2b0b 100644 --- a/openstackclient/tests/network/v2/test_ip_availability.py +++ b/openstackclient/tests/network/v2/test_ip_availability.py @@ -82,8 +82,10 @@ class TestListIPAvailability(TestIPAvailability): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) + filters = {'ip_version': 4} - self.network.network_ip_availabilities.assert_called_once_with() + self.network.network_ip_availabilities.assert_called_once_with( + **filters) self.assertEqual(self.columns, columns) self.assertEqual(self.data, list(data)) @@ -116,7 +118,8 @@ class TestListIPAvailability(TestIPAvailability): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) - filters = {'tenant_id': identity_fakes.project_id} + filters = {'tenant_id': identity_fakes.project_id, + 'ip_version': 4} self.network.network_ip_availabilities.assert_called_once_with( **filters) diff --git a/openstackclient/tests/network/v2/test_security_group.py b/openstackclient/tests/network/v2/test_security_group.py index 213367a4..b0c14985 100644 --- a/openstackclient/tests/network/v2/test_security_group.py +++ b/openstackclient/tests/network/v2/test_security_group.py @@ -13,6 +13,9 @@ import copy import mock +from mock import call + +from osc_lib import exceptions from openstackclient.network.v2 import security_group from openstackclient.tests.compute.v2 import fakes as compute_fakes @@ -227,42 +230,93 @@ class TestCreateSecurityGroupCompute(TestSecurityGroupCompute): class TestDeleteSecurityGroupNetwork(TestSecurityGroupNetwork): - # The security group to be deleted. - _security_group = \ - network_fakes.FakeSecurityGroup.create_one_security_group() + # The security groups to be deleted. + _security_groups = \ + network_fakes.FakeSecurityGroup.create_security_groups() 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) + self.network.find_security_group = ( + network_fakes.FakeSecurityGroup.get_security_groups( + self._security_groups) + ) # 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, + self._security_groups[0].name, ] verifylist = [ - ('group', self._security_group.name), + ('group', [self._security_groups[0].name]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) self.network.delete_security_group.assert_called_once_with( - self._security_group) + self._security_groups[0]) + self.assertIsNone(result) + + def test_multi_security_groups_delete(self): + arglist = [] + verifylist = [] + + for s in self._security_groups: + arglist.append(s.name) + verifylist = [ + ('group', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for s in self._security_groups: + calls.append(call(s)) + self.network.delete_security_group.assert_has_calls(calls) self.assertIsNone(result) + def test_multi_security_groups_delete_with_exception(self): + arglist = [ + self._security_groups[0].name, + 'unexist_security_group', + ] + verifylist = [ + ('group', + [self._security_groups[0].name, 'unexist_security_group']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self._security_groups[0], exceptions.CommandError] + self.network.find_security_group = ( + mock.MagicMock(side_effect=find_mock_result) + ) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 group failed to delete.', str(e)) + + self.network.find_security_group.assert_any_call( + self._security_groups[0].name, ignore_missing=False) + self.network.find_security_group.assert_any_call( + 'unexist_security_group', ignore_missing=False) + self.network.delete_security_group.assert_called_once_with( + self._security_groups[0] + ) + class TestDeleteSecurityGroupCompute(TestSecurityGroupCompute): - # The security group to be deleted. - _security_group = \ - compute_fakes.FakeSecurityGroup.create_one_security_group() + # The security groups to be deleted. + _security_groups = \ + compute_fakes.FakeSecurityGroup.create_security_groups() def setUp(self): super(TestDeleteSecurityGroupCompute, self).setUp() @@ -271,27 +325,80 @@ class TestDeleteSecurityGroupCompute(TestSecurityGroupCompute): self.compute.security_groups.delete = mock.Mock(return_value=None) - self.compute.security_groups.get = mock.Mock( - return_value=self._security_group) + self.compute.security_groups.get = ( + compute_fakes.FakeSecurityGroup.get_security_groups( + self._security_groups) + ) # Get the command object to test self.cmd = security_group.DeleteSecurityGroup(self.app, None) def test_security_group_delete(self): arglist = [ - self._security_group.name, + self._security_groups[0].id, ] verifylist = [ - ('group', self._security_group.name), + ('group', [self._security_groups[0].id]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) self.compute.security_groups.delete.assert_called_once_with( - self._security_group.id) + self._security_groups[0].id) + self.assertIsNone(result) + + def test_multi_security_groups_delete(self): + arglist = [] + verifylist = [] + + for s in self._security_groups: + arglist.append(s.id) + verifylist = [ + ('group', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for s in self._security_groups: + calls.append(call(s.id)) + self.compute.security_groups.delete.assert_has_calls(calls) self.assertIsNone(result) + def test_multi_security_groups_delete_with_exception(self): + arglist = [ + self._security_groups[0].id, + 'unexist_security_group', + ] + verifylist = [ + ('group', + [self._security_groups[0].id, 'unexist_security_group']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self._security_groups[0], exceptions.CommandError] + self.compute.security_groups.get = ( + mock.MagicMock(side_effect=find_mock_result) + ) + self.compute.security_groups.find.side_effect = ( + exceptions.NotFound(None)) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 group failed to delete.', str(e)) + + self.compute.security_groups.get.assert_any_call( + self._security_groups[0].id) + self.compute.security_groups.get.assert_any_call( + 'unexist_security_group') + self.compute.security_groups.delete.assert_called_once_with( + self._security_groups[0].id + ) + class TestListSecurityGroupNetwork(TestSecurityGroupNetwork): diff --git a/openstackclient/tests/network/v2/test_security_group_rule.py b/openstackclient/tests/network/v2/test_security_group_rule.py index b1f2209d..b2862679 100644 --- a/openstackclient/tests/network/v2/test_security_group_rule.py +++ b/openstackclient/tests/network/v2/test_security_group_rule.py @@ -13,6 +13,7 @@ import copy import mock +from mock import call from osc_lib import exceptions @@ -668,17 +669,20 @@ class TestCreateSecurityGroupRuleCompute(TestSecurityGroupRuleCompute): class TestDeleteSecurityGroupRuleNetwork(TestSecurityGroupRuleNetwork): - # The security group rule to be deleted. - _security_group_rule = \ - network_fakes.FakeSecurityGroupRule.create_one_security_group_rule() + # The security group rules to be deleted. + _security_group_rules = \ + network_fakes.FakeSecurityGroupRule.create_security_group_rules( + count=2) 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) + self.network.find_security_group_rule = ( + network_fakes.FakeSecurityGroupRule.get_security_group_rules( + self._security_group_rules) + ) # Get the command object to test self.cmd = security_group_rule.DeleteSecurityGroupRule( @@ -686,25 +690,76 @@ class TestDeleteSecurityGroupRuleNetwork(TestSecurityGroupRuleNetwork): def test_security_group_rule_delete(self): arglist = [ - self._security_group_rule.id, + self._security_group_rules[0].id, ] verifylist = [ - ('rule', self._security_group_rule.id), + ('rule', [self._security_group_rules[0].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_once_with( - self._security_group_rule) + self._security_group_rules[0]) + self.assertIsNone(result) + + def test_multi_security_group_rules_delete(self): + arglist = [] + verifylist = [] + + for s in self._security_group_rules: + arglist.append(s.id) + verifylist = [ + ('rule', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for s in self._security_group_rules: + calls.append(call(s)) + self.network.delete_security_group_rule.assert_has_calls(calls) self.assertIsNone(result) + def test_multi_security_group_rules_delete_with_exception(self): + arglist = [ + self._security_group_rules[0].id, + 'unexist_rule', + ] + verifylist = [ + ('rule', + [self._security_group_rules[0].id, 'unexist_rule']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [ + self._security_group_rules[0], exceptions.CommandError] + self.network.find_security_group_rule = ( + mock.MagicMock(side_effect=find_mock_result) + ) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 rule failed to delete.', str(e)) + + self.network.find_security_group_rule.assert_any_call( + self._security_group_rules[0].id, ignore_missing=False) + self.network.find_security_group_rule.assert_any_call( + 'unexist_rule', ignore_missing=False) + self.network.delete_security_group_rule.assert_called_once_with( + self._security_group_rules[0] + ) + class TestDeleteSecurityGroupRuleCompute(TestSecurityGroupRuleCompute): # The security group rule to be deleted. - _security_group_rule = \ - compute_fakes.FakeSecurityGroupRule.create_one_security_group_rule() + _security_group_rules = \ + compute_fakes.FakeSecurityGroupRule.create_security_group_rules( + count=2) def setUp(self): super(TestDeleteSecurityGroupRuleCompute, self).setUp() @@ -716,19 +771,65 @@ class TestDeleteSecurityGroupRuleCompute(TestSecurityGroupRuleCompute): def test_security_group_rule_delete(self): arglist = [ - self._security_group_rule.id, + self._security_group_rules[0].id, ] verifylist = [ - ('rule', self._security_group_rule.id), + ('rule', [self._security_group_rules[0].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_once_with( - self._security_group_rule.id) + self._security_group_rules[0].id) + self.assertIsNone(result) + + def test_multi_security_group_rules_delete(self): + arglist = [] + verifylist = [] + + for s in self._security_group_rules: + arglist.append(s.id) + verifylist = [ + ('rule', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for s in self._security_group_rules: + calls.append(call(s.id)) + self.compute.security_group_rules.delete.assert_has_calls(calls) self.assertIsNone(result) + def test_multi_security_group_rules_delete_with_exception(self): + arglist = [ + self._security_group_rules[0].id, + 'unexist_rule', + ] + verifylist = [ + ('rule', + [self._security_group_rules[0].id, 'unexist_rule']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [None, exceptions.CommandError] + self.compute.security_group_rules.delete = ( + mock.MagicMock(side_effect=find_mock_result) + ) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 rule failed to delete.', str(e)) + + self.compute.security_group_rules.delete.assert_any_call( + self._security_group_rules[0].id) + self.compute.security_group_rules.delete.assert_any_call( + 'unexist_rule') + class TestListSecurityGroupRuleNetwork(TestSecurityGroupRuleNetwork): diff --git a/openstackclient/tests/network/v2/test_subnet.py b/openstackclient/tests/network/v2/test_subnet.py index de7e1821..a57a0308 100644 --- a/openstackclient/tests/network/v2/test_subnet.py +++ b/openstackclient/tests/network/v2/test_subnet.py @@ -13,7 +13,9 @@ import copy import mock +from mock import call +from osc_lib import exceptions from osc_lib import utils from openstackclient.network.v2 import subnet as subnet_v2 @@ -361,32 +363,82 @@ class TestCreateSubnet(TestSubnet): class TestDeleteSubnet(TestSubnet): - # The subnet to delete. - _subnet = network_fakes.FakeSubnet.create_one_subnet() + # The subnets to delete. + _subnets = network_fakes.FakeSubnet.create_subnets(count=2) def setUp(self): super(TestDeleteSubnet, self).setUp() self.network.delete_subnet = mock.Mock(return_value=None) - self.network.find_subnet = mock.Mock(return_value=self._subnet) + self.network.find_subnet = ( + network_fakes.FakeSubnet.get_subnets(self._subnets)) # Get the command object to test self.cmd = subnet_v2.DeleteSubnet(self.app, self.namespace) - def test_delete(self): + def test_subnet_delete(self): arglist = [ - self._subnet.name, + self._subnets[0].name, ] verifylist = [ - ('subnet', self._subnet.name), + ('subnet', [self._subnets[0].name]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.network.delete_subnet.assert_called_once_with(self._subnet) + self.network.delete_subnet.assert_called_once_with(self._subnets[0]) self.assertIsNone(result) + def test_multi_subnets_delete(self): + arglist = [] + verifylist = [] + + for s in self._subnets: + arglist.append(s.name) + verifylist = [ + ('subnet', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for s in self._subnets: + calls.append(call(s)) + self.network.delete_subnet.assert_has_calls(calls) + self.assertIsNone(result) + + def test_multi_subnets_delete_with_exception(self): + arglist = [ + self._subnets[0].name, + 'unexist_subnet', + ] + verifylist = [ + ('subnet', + [self._subnets[0].name, 'unexist_subnet']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self._subnets[0], exceptions.CommandError] + self.network.find_subnet = ( + mock.MagicMock(side_effect=find_mock_result) + ) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 subnets failed to delete.', str(e)) + + self.network.find_subnet.assert_any_call( + self._subnets[0].name, ignore_missing=False) + self.network.find_subnet.assert_any_call( + 'unexist_subnet', ignore_missing=False) + self.network.delete_subnet.assert_called_once_with( + self._subnets[0] + ) + class TestListSubnet(TestSubnet): # The subnets going to be listed up. diff --git a/openstackclient/tests/network/v2/test_subnet_pool.py b/openstackclient/tests/network/v2/test_subnet_pool.py index 10ef76d8..7a96b30f 100644 --- a/openstackclient/tests/network/v2/test_subnet_pool.py +++ b/openstackclient/tests/network/v2/test_subnet_pool.py @@ -14,7 +14,9 @@ import argparse import copy import mock +from mock import call +from osc_lib import exceptions from osc_lib import utils from openstackclient.network.v2 import subnet_pool @@ -263,36 +265,85 @@ class TestCreateSubnetPool(TestSubnetPool): class TestDeleteSubnetPool(TestSubnetPool): - # The subnet pool to delete. - _subnet_pool = network_fakes.FakeSubnetPool.create_one_subnet_pool() + # The subnet pools to delete. + _subnet_pools = network_fakes.FakeSubnetPool.create_subnet_pools(count=2) def setUp(self): super(TestDeleteSubnetPool, self).setUp() self.network.delete_subnet_pool = mock.Mock(return_value=None) - self.network.find_subnet_pool = mock.Mock( - return_value=self._subnet_pool + self.network.find_subnet_pool = ( + network_fakes.FakeSubnetPool.get_subnet_pools(self._subnet_pools) ) # Get the command object to test self.cmd = subnet_pool.DeleteSubnetPool(self.app, self.namespace) - def test_delete(self): + def test_subnet_pool_delete(self): arglist = [ - self._subnet_pool.name, + self._subnet_pools[0].name, ] verifylist = [ - ('subnet_pool', self._subnet_pool.name), + ('subnet_pool', [self._subnet_pools[0].name]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) self.network.delete_subnet_pool.assert_called_once_with( - self._subnet_pool) + self._subnet_pools[0]) + self.assertIsNone(result) + + def test_multi_subnet_pools_delete(self): + arglist = [] + verifylist = [] + + for s in self._subnet_pools: + arglist.append(s.name) + verifylist = [ + ('subnet_pool', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for s in self._subnet_pools: + calls.append(call(s)) + self.network.delete_subnet_pool.assert_has_calls(calls) self.assertIsNone(result) + def test_multi_subnet_pools_delete_with_exception(self): + arglist = [ + self._subnet_pools[0].name, + 'unexist_subnet_pool', + ] + verifylist = [ + ('subnet_pool', + [self._subnet_pools[0].name, 'unexist_subnet_pool']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self._subnet_pools[0], exceptions.CommandError] + self.network.find_subnet_pool = ( + mock.MagicMock(side_effect=find_mock_result) + ) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 subnet pools failed to delete.', str(e)) + + self.network.find_subnet_pool.assert_any_call( + self._subnet_pools[0].name, ignore_missing=False) + self.network.find_subnet_pool.assert_any_call( + 'unexist_subnet_pool', ignore_missing=False) + self.network.delete_subnet_pool.assert_called_once_with( + self._subnet_pools[0] + ) + class TestListSubnetPool(TestSubnetPool): # The subnet pools going to be listed up. diff --git a/openstackclient/tests/test_shell.py b/openstackclient/tests/test_shell.py index 90454fc2..7d0bbd12 100644 --- a/openstackclient/tests/test_shell.py +++ b/openstackclient/tests/test_shell.py @@ -1,4 +1,4 @@ -# Copyright 2012-2013 OpenStack, LLC. +# Copyright 2012-2013 OpenStack Foundation # # 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 diff --git a/openstackclient/tests/volume/v1/test_volume.py b/openstackclient/tests/volume/v1/test_volume.py index e4f51bb5..380bc632 100644 --- a/openstackclient/tests/volume/v1/test_volume.py +++ b/openstackclient/tests/volume/v1/test_volume.py @@ -14,6 +14,7 @@ # import copy +import mock from openstackclient.tests import fakes from openstackclient.tests.identity.v2_0 import fakes as identity_fakes @@ -656,7 +657,8 @@ class TestVolumeSet(TestVolume): ) self.assertIsNone(result) - def test_volume_set_size_smaller(self): + @mock.patch.object(volume.LOG, 'error') + def test_volume_set_size_smaller(self, mock_log_error): arglist = [ '--size', '100', volume_fakes.volume_name, @@ -672,12 +674,13 @@ class TestVolumeSet(TestVolume): result = self.cmd.take_action(parsed_args) - self.assertEqual("New size must be greater than %s GB" % - volume_fakes.volume_size, - self.app.log.messages.get('error')) + mock_log_error.assert_called_with("New size must be greater " + "than %s GB", + volume_fakes.volume_size) self.assertIsNone(result) - def test_volume_set_size_not_available(self): + @mock.patch.object(volume.LOG, 'error') + def test_volume_set_size_not_available(self, mock_log_error): self.volumes_mock.get.return_value.status = 'error' arglist = [ '--size', '130', @@ -694,9 +697,9 @@ class TestVolumeSet(TestVolume): result = self.cmd.take_action(parsed_args) - self.assertEqual("Volume is in %s state, it must be available before " - "size can be extended" % 'error', - self.app.log.messages.get('error')) + mock_log_error.assert_called_with("Volume is in %s state, it must be " + "available before size can be " + "extended", 'error') self.assertIsNone(result) def test_volume_set_property(self): |