# Copyright 2014: Mirantis Inc. # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import copy from boto import exception as boto_exception import ddt import mock from neutronclient.common import exceptions as neutron_exceptions from novaclient import exceptions as nova_exc from watcherclient.common.apiclient import exceptions as watcher_exceptions from rally_openstack.cleanup import resources from tests.unit import test BASE = "rally_openstack.cleanup.resources" GLANCE_V2_PATH = ("rally_openstack.services.image.glance_v2." "GlanceV2Service") class SynchronizedDeletionTestCase(test.TestCase): def test_is_deleted(self): self.assertTrue(resources.SynchronizedDeletion().is_deleted()) class QuotaMixinTestCase(test.TestCase): @mock.patch("%s.identity.Identity" % BASE) def test_list(self, mock_identity): quota = resources.QuotaMixin() quota.tenant_uuid = None quota.user = mock.MagicMock() self.assertEqual([], quota.list()) self.assertFalse(mock_identity.called) quota.tenant_uuid = mock.MagicMock() self.assertEqual([mock_identity.return_value.get_project.return_value], quota.list()) mock_identity.assert_called_once_with(quota.user) class MagnumMixinTestCase(test.TestCase): def test_id(self): magnum = resources.MagnumMixin() magnum._service = "magnum" magnum.raw_resource = mock.MagicMock() self.assertEqual(magnum.raw_resource.uuid, magnum.id()) def test_list(self): magnum = resources.MagnumMixin() magnum._service = "magnum" some_resources = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock()] magnum._manager = mock.MagicMock() magnum._manager.return_value.list.side_effect = ( some_resources[:2], some_resources[2:4], []) self.assertEqual(some_resources, magnum.list()) self.assertEqual( [mock.call(marker=None), mock.call(marker=some_resources[1].uuid), mock.call(marker=some_resources[3].uuid)], magnum._manager.return_value.list.call_args_list) class NovaServerTestCase(test.TestCase): def test_list(self): server = resources.NovaServer() server._manager = mock.MagicMock() server.list() server._manager.return_value.list.assert_called_once_with(limit=-1) def test_delete(self): server = resources.NovaServer() server.raw_resource = mock.Mock() server._manager = mock.Mock() server.delete() server._manager.return_value.delete.assert_called_once_with( server.raw_resource.id) def test_delete_locked(self): server = resources.NovaServer() server.raw_resource = mock.Mock() setattr(server.raw_resource, "OS-EXT-STS:locked", True) server._manager = mock.Mock() server.delete() server.raw_resource.unlock.assert_called_once_with() server._manager.return_value.delete.assert_called_once_with( server.raw_resource.id) class NovaFlavorsTestCase(test.TestCase): @mock.patch("%s.base.ResourceManager._manager" % BASE) def test_is_deleted(self, mock_resource_manager__manager): exc = nova_exc.NotFound(404) mock_resource_manager__manager().get.side_effect = exc flavor = resources.NovaFlavors() flavor.raw_resource = mock.MagicMock() self.assertTrue(flavor.is_deleted()) @mock.patch("%s.base.ResourceManager._manager" % BASE) def test_is_deleted_fail(self, mock_resource_manager__manager): mock_resource_manager__manager().get.side_effect = TypeError() flavor = resources.NovaFlavors() flavor.raw_resource = mock.MagicMock() self.assertRaises(TypeError, flavor.is_deleted) class NovaServerGroupsTestCase(test.TestCase): @mock.patch("%s.base.ResourceManager._manager" % BASE) @mock.patch("rally.common.utils.name_matches_object") def test_list(self, mock_name_matches_object, mock_resource_manager__manager): server_groups = [mock.MagicMock(name="rally_foo1"), mock.MagicMock(name="rally_foo2"), mock.MagicMock(name="foo3")] mock_name_matches_object.side_effect = [False, True, True] mock_resource_manager__manager().list.return_value = server_groups self.assertEqual(server_groups, resources.NovaServerGroups().list()) class EC2MixinTestCase(test.TestCase): def get_ec2_mixin(self): ec2 = resources.EC2Mixin() ec2._service = "ec2" return ec2 def test__manager(self): ec2 = self.get_ec2_mixin() ec2.user = mock.MagicMock() self.assertEqual(ec2.user.ec2.return_value, ec2._manager()) class EC2ServerTestCase(test.TestCase): @mock.patch("%s.EC2Server._manager" % BASE) def test_is_deleted(self, mock_ec2_server__manager): raw_res1 = mock.MagicMock(state="terminated") raw_res2 = mock.MagicMock(state="terminated") resource = mock.MagicMock(id="test_id") manager = resources.EC2Server(resource=resource) mock_ec2_server__manager().get_only_instances.return_value = [raw_res1] self.assertTrue(manager.is_deleted()) raw_res1.state = "running" self.assertFalse(manager.is_deleted()) mock_ec2_server__manager().get_only_instances.return_value = [ raw_res1, raw_res2] self.assertFalse(manager.is_deleted()) raw_res1.state = "terminated" self.assertTrue(manager.is_deleted()) mock_ec2_server__manager().get_only_instances.return_value = [] self.assertTrue(manager.is_deleted()) @mock.patch("%s.EC2Server._manager" % BASE) def test_is_deleted_exceptions(self, mock_ec2_server__manager): mock_ec2_server__manager.side_effect = [ boto_exception.EC2ResponseError( status="fake", reason="fake", body={"Error": {"Code": "fake_code"}}), boto_exception.EC2ResponseError( status="fake", reason="fake", body={"Error": {"Code": "InvalidInstanceID.NotFound"}}) ] manager = resources.EC2Server(resource=mock.MagicMock()) self.assertFalse(manager.is_deleted()) self.assertTrue(manager.is_deleted()) @mock.patch("%s.EC2Server._manager" % BASE) def test_delete(self, mock_ec2_server__manager): resource = mock.MagicMock(id="test_id") manager = resources.EC2Server(resource=resource) manager.delete() mock_ec2_server__manager().terminate_instances.assert_called_once_with( instance_ids=["test_id"]) @mock.patch("%s.EC2Server._manager" % BASE) def test_list(self, mock_ec2_server__manager): manager = resources.EC2Server() mock_ec2_server__manager().get_only_instances.return_value = [ "a", "b", "c"] self.assertEqual(["a", "b", "c"], manager.list()) class NeutronMixinTestCase(test.TestCase): def get_neutron_mixin(self): neut = resources.NeutronMixin() neut._service = "neutron" return neut def test_manager(self): neut = self.get_neutron_mixin() neut.user = mock.MagicMock() self.assertEqual(neut.user.neutron.return_value, neut._manager()) @mock.patch("%s.NeutronMixin._manager" % BASE) def test_supports_extension(self, mock__manager): mock__manager().list_extensions.return_value = { "extensions": [{"alias": "foo"}, {"alias": "bar"}] } neut = self.get_neutron_mixin() self.assertTrue(neut.supports_extension("foo")) self.assertTrue(neut.supports_extension("bar")) self.assertFalse(neut.supports_extension("foobar")) def test_id(self): neut = self.get_neutron_mixin() neut.raw_resource = {"id": "test"} self.assertEqual("test", neut.id()) def test_name(self): neutron = self.get_neutron_mixin() neutron.raw_resource = {"id": "test_id", "name": "test_name"} self.assertEqual("test_name", neutron.name()) def test_delete(self): neut = self.get_neutron_mixin() neut.user = mock.MagicMock() neut._resource = "some_resource" neut.raw_resource = {"id": "42"} neut.delete() neut.user.neutron().delete_some_resource.assert_called_once_with("42") def test_list(self): neut = self.get_neutron_mixin() neut.user = mock.MagicMock() neut._resource = "some_resource" neut.tenant_uuid = "user_tenant" some_resources = [{"tenant_id": neut.tenant_uuid}, {"tenant_id": "a"}] neut.user.neutron().list_some_resources.return_value = { "some_resources": some_resources } self.assertEqual([some_resources[0]], list(neut.list())) neut.user.neutron().list_some_resources.assert_called_once_with( tenant_id=neut.tenant_uuid) class NeutronLbaasV1MixinTestCase(test.TestCase): def get_neutron_lbaasv1_mixin(self, extensions=None): if extensions is None: extensions = [] neut = resources.NeutronLbaasV1Mixin() neut._service = "neutron" neut._resource = "some_resource" neut._manager = mock.Mock() neut._manager().list_extensions.return_value = { "extensions": [{"alias": ext} for ext in extensions] } return neut def test_list_lbaas_available(self): neut = self.get_neutron_lbaasv1_mixin(extensions=["lbaas"]) neut.tenant_uuid = "user_tenant" some_resources = [{"tenant_id": neut.tenant_uuid}, {"tenant_id": "a"}] neut._manager().list_some_resources.return_value = { "some_resources": some_resources } self.assertEqual([some_resources[0]], list(neut.list())) neut._manager().list_some_resources.assert_called_once_with( tenant_id=neut.tenant_uuid) def test_list_lbaas_unavailable(self): neut = self.get_neutron_lbaasv1_mixin() self.assertEqual([], list(neut.list())) self.assertFalse(neut._manager().list_some_resources.called) class NeutronLbaasV2MixinTestCase(test.TestCase): def get_neutron_lbaasv2_mixin(self, extensions=None): if extensions is None: extensions = [] neut = resources.NeutronLbaasV2Mixin() neut._service = "neutron" neut._resource = "some_resource" neut._manager = mock.Mock() neut._manager().list_extensions.return_value = { "extensions": [{"alias": ext} for ext in extensions] } return neut def test_list_lbaasv2_available(self): neut = self.get_neutron_lbaasv2_mixin(extensions=["lbaasv2"]) neut.tenant_uuid = "user_tenant" some_resources = [{"tenant_id": neut.tenant_uuid}, {"tenant_id": "a"}] neut._manager().list_some_resources.return_value = { "some_resources": some_resources } self.assertEqual([some_resources[0]], list(neut.list())) neut._manager().list_some_resources.assert_called_once_with( tenant_id=neut.tenant_uuid) def test_list_lbaasv2_unavailable(self): neut = self.get_neutron_lbaasv2_mixin() self.assertEqual([], list(neut.list())) self.assertFalse(neut._manager().list_some_resources.called) class NeutronV2LoadbalancerTestCase(test.TestCase): def get_neutron_lbaasv2_lb(self): neutron_lb = resources.NeutronV2Loadbalancer() neutron_lb.raw_resource = {"id": "1", "name": "s_rally"} neutron_lb._manager = mock.Mock() return neutron_lb def test_is_deleted_true(self): from neutronclient.common import exceptions as n_exceptions neutron_lb = self.get_neutron_lbaasv2_lb() neutron_lb._manager().show_loadbalancer.side_effect = ( n_exceptions.NotFound) self.assertTrue(neutron_lb.is_deleted()) neutron_lb._manager().show_loadbalancer.assert_called_once_with( neutron_lb.id()) def test_is_deleted_false(self): from neutronclient.common import exceptions as n_exceptions neutron_lb = self.get_neutron_lbaasv2_lb() neutron_lb._manager().show_loadbalancer.return_value = ( neutron_lb.raw_resource) self.assertFalse(neutron_lb.is_deleted()) neutron_lb._manager().show_loadbalancer.assert_called_once_with( neutron_lb.id()) neutron_lb._manager().show_loadbalancer.reset_mock() neutron_lb._manager().show_loadbalancer.side_effect = ( n_exceptions.Forbidden) self.assertFalse(neutron_lb.is_deleted()) neutron_lb._manager().show_loadbalancer.assert_called_once_with( neutron_lb.id()) class NeutronBgpvpnTestCase(test.TestCase): def get_neutron_bgpvpn_mixin(self, extensions=None): if extensions is None: extensions = [] admin = mock.Mock() neut = resources.NeutronBgpvpn(admin=admin) neut._manager = mock.Mock() neut._manager().list_extensions.return_value = { "extensions": [{"alias": ext} for ext in extensions] } return neut def test_list_user(self): neut = self.get_neutron_bgpvpn_mixin(extensions=["bgpvpn"]) user_bgpvpns = {"bgpvpns": [{"tenant_id": "foo", "id": "bgpvpn_id"}]} neut._manager().list_bgpvpns.return_value = user_bgpvpns bgpvpns_list = neut.list() self.assertEqual("bgpvpn", neut._resource) neut._manager().list_bgpvpns.assert_called_once_with() self.assertEqual(bgpvpns_list, user_bgpvpns["bgpvpns"]) def test_list_admin(self): neut = self.get_neutron_bgpvpn_mixin(extensions=["bgpvpn"]) admin_bgpvpns = {"bgpvpns": [{"tenant_id": "foo", "id": "bgpvpn_id"}]} neut._manager().list_bgpvpns.return_value = admin_bgpvpns self.assertEqual("bgpvpn", neut._resource) self.assertEqual(neut.list(), admin_bgpvpns["bgpvpns"]) class NeutronFloatingIPTestCase(test.TestCase): def test_name(self): fips = resources.NeutronFloatingIP({"name": "foo", "description": "OoO"}) self.assertEqual(fips.name(), "OoO") def test_list(self): fips = {"floatingips": [{"tenant_id": "foo", "id": "foo"}]} user = mock.MagicMock() user.neutron.return_value.list_floatingips.return_value = fips self.assertEqual(fips["floatingips"], list( resources.NeutronFloatingIP(user=user, tenant_uuid="foo").list())) user.neutron.return_value.list_floatingips.assert_called_once_with( tenant_id="foo") class NeutronPortTestCase(test.TestCase): def test_delete(self): raw_res = {"device_owner": "abbabaab", "id": "some_id"} user = mock.MagicMock() resources.NeutronPort(resource=raw_res, user=user).delete() user.neutron().delete_port.assert_called_once_with(raw_res["id"]) def test_delete_port_raise_exception(self): raw_res = {"device_owner": "abbabaab", "id": "some_id"} user = mock.MagicMock() user.neutron().delete_port.side_effect = ( neutron_exceptions.PortNotFoundClient) resources.NeutronPort(resource=raw_res, user=user).delete() user.neutron().delete_port.assert_called_once_with(raw_res["id"]) def test_delete_port_device_owner(self): raw_res = { "device_owner": "network:router_interface", "id": "some_id", "device_id": "dev_id" } user = mock.MagicMock() resources.NeutronPort(resource=raw_res, user=user).delete() user.neutron().remove_interface_router.assert_called_once_with( raw_res["device_id"], {"port_id": raw_res["id"]}) def test_name(self): raw_res = { "id": "some_id", "device_id": "dev_id", } # automatically created or manually created port. No name field self.assertEqual( resources.NeutronPort(resource=raw_res, user=mock.MagicMock()).name(), "") raw_res["name"] = "foo" self.assertEqual("foo", resources.NeutronPort( resource=raw_res, user=mock.MagicMock()).name()) raw_res["parent_name"] = "bar" self.assertEqual("bar", resources.NeutronPort( resource=raw_res, user=mock.MagicMock()).name()) del raw_res["name"] self.assertEqual("bar", resources.NeutronPort( resource=raw_res, user=mock.MagicMock()).name()) def test_list(self): tenant_uuid = "uuuu-uuuu-iiii-dddd" ports = [ # the case when 'name' is present, so 'device_owner' field is not # required {"tenant_id": tenant_uuid, "id": "id1", "name": "foo"}, # 3 different cases when router_interface is an owner {"tenant_id": tenant_uuid, "id": "id2", "device_owner": "network:router_interface", "device_id": "router-1"}, {"tenant_id": tenant_uuid, "id": "id3", "device_owner": "network:router_interface_distributed", "device_id": "router-1"}, {"tenant_id": tenant_uuid, "id": "id4", "device_owner": "network:ha_router_replicated_interface", "device_id": "router-2"}, # the case when gateway router is an owner {"tenant_id": tenant_uuid, "id": "id5", "device_owner": "network:router_gateway", "device_id": "router-3"}, # the case when gateway router is an owner, but device_id is # invalid {"tenant_id": tenant_uuid, "id": "id6", "device_owner": "network:router_gateway", "device_id": "aaaa"}, # the case when port was auto-created with floating-ip {"tenant_id": tenant_uuid, "id": "id7", "device_owner": "network:dhcp", "device_id": "asdasdasd"}, # the case when port is from another tenant {"tenant_id": "wrong tenant", "id": "id8", "name": "foo"}, # WTF port without any parent and name {"tenant_id": tenant_uuid, "id": "id9", "device_owner": ""}, ] routers = [ {"id": "router-1", "name": "Router-1", "tenant_id": tenant_uuid}, {"id": "router-2", "name": "Router-2", "tenant_id": tenant_uuid}, {"id": "router-3", "name": "Router-3", "tenant_id": tenant_uuid}, {"id": "router-4", "name": "Router-4", "tenant_id": tenant_uuid}, {"id": "router-5", "name": "Router-5", "tenant_id": tenant_uuid}, ] expected_ports = [] for port in ports: if port["tenant_id"] == tenant_uuid: expected_ports.append(copy.deepcopy(port)) if ("device_id" in port and port["device_id"].startswith("router")): expected_ports[-1]["parent_name"] = [ r for r in routers if r["id"] == port["device_id"]][0]["name"] class FakeNeutronClient(object): list_ports = mock.Mock() list_routers = mock.Mock() neutron = FakeNeutronClient neutron.list_ports.return_value = {"ports": ports} neutron.list_routers.return_value = {"routers": routers} user = mock.Mock(neutron=neutron) self.assertEqual(expected_ports, resources.NeutronPort( user=user, tenant_uuid=tenant_uuid).list()) neutron.list_ports.assert_called_once_with() neutron.list_routers.assert_called_once_with() @ddt.ddt class NeutronSecurityGroupTestCase(test.TestCase): @ddt.data( {"admin": mock.Mock(), "admin_required": True}, {"admin": None, "admin_required": False}) @ddt.unpack def test_list(self, admin, admin_required): sg_list = [{"tenant_id": "user_tenant", "name": "default"}, {"tenant_id": "user_tenant", "name": "foo_sg"}] neut = resources.NeutronSecurityGroup() neut.user = mock.MagicMock() neut._resource = "security_group" neut.tenant_uuid = "user_tenant" neut.user.neutron().list_security_groups.return_value = { "security_groups": sg_list } expected_result = [sg_list[1]] self.assertEqual(expected_result, list(neut.list())) neut.user.neutron().list_security_groups.assert_called_once_with( tenant_id=neut.tenant_uuid) class NeutronQuotaTestCase(test.TestCase): def test_delete(self): admin = mock.MagicMock() resources.NeutronQuota(admin=admin, tenant_uuid="fake").delete() admin.neutron.return_value.delete_quota.assert_called_once_with("fake") @ddt.ddt class GlanceImageTestCase(test.TestCase): @mock.patch("rally_openstack.services.image.image.Image") def test__client_admin(self, mock_image): admin = mock.Mock() glance = resources.GlanceImage(admin=admin) client = glance._client() mock_image.assert_called_once_with(admin) self.assertEqual(client, mock_image.return_value) @mock.patch("rally_openstack.services.image.image.Image") def test__client_user(self, mock_image): user = mock.Mock() glance = resources.GlanceImage(user=user) wrapper = glance._client() mock_image.assert_called_once_with(user) self.assertEqual(wrapper, mock_image.return_value) @mock.patch("rally_openstack.services.image.image.Image") def test__client_admin_preferred(self, mock_image): admin = mock.Mock() user = mock.Mock() glance = resources.GlanceImage(admin=admin, user=user) client = glance._client() mock_image.assert_called_once_with(admin) self.assertEqual(client, mock_image.return_value) def test_list(self): glance = resources.GlanceImage() glance._client = mock.Mock() list_images = glance._client.return_value.list_images list_images.side_effect = ( ["active-image1", "active-image2"], ["deactivated-image1"]) glance.tenant_uuid = mock.Mock() self.assertEqual( glance.list(), ["active-image1", "active-image2", "deactivated-image1"]) list_images.assert_has_calls([ mock.call(owner=glance.tenant_uuid), mock.call(status="deactivated", owner=glance.tenant_uuid)]) def test_delete(self): glance = resources.GlanceImage() glance._client = mock.Mock() glance._wrapper = mock.Mock() glance.raw_resource = mock.Mock() client = glance._client.return_value deleted_image = mock.Mock(status="DELETED") client.get_image.side_effect = [glance.raw_resource, deleted_image] glance.delete() client.delete_image.assert_called_once_with(glance.raw_resource.id) self.assertFalse(client.reactivate_image.called) @mock.patch("%s.reactivate_image" % GLANCE_V2_PATH) def test_delete_deactivated_image(self, mock_reactivate_image): glance = resources.GlanceImage() glance._client = mock.Mock() glance._wrapper = mock.Mock() glance.raw_resource = mock.Mock(status="deactivated") client = glance._client.return_value deleted_image = mock.Mock(status="DELETED") client.get_image.side_effect = [glance.raw_resource, deleted_image] glance.delete() mock_reactivate_image.assert_called_once_with(glance.raw_resource.id) client.delete_image.assert_called_once_with(glance.raw_resource.id) class CeilometerTestCase(test.TestCase): def test_id(self): ceil = resources.CeilometerAlarms() ceil.raw_resource = mock.MagicMock() self.assertEqual(ceil.raw_resource.alarm_id, ceil.id()) @mock.patch("%s.CeilometerAlarms._manager" % BASE) def test_list(self, mock_ceilometer_alarms__manager): ceil = resources.CeilometerAlarms() ceil.tenant_uuid = mock.MagicMock() mock_ceilometer_alarms__manager().list.return_value = ["a", "b", "c"] mock_ceilometer_alarms__manager.reset_mock() self.assertEqual(["a", "b", "c"], ceil.list()) mock_ceilometer_alarms__manager().list.assert_called_once_with( q=[{"field": "project_id", "op": "eq", "value": ceil.tenant_uuid}]) class ZaqarQueuesTestCase(test.TestCase): def test_list(self): user = mock.Mock() zaqar = resources.ZaqarQueues(user=user) zaqar.list() user.zaqar().queues.assert_called_once_with() class KeystoneMixinTestCase(test.TestCase): def test_is_deleted(self): self.assertTrue(resources.KeystoneMixin().is_deleted()) def get_keystone_mixin(self): kmixin = resources.KeystoneMixin() kmixin._service = "keystone" return kmixin @mock.patch("%s.identity" % BASE) def test_manager(self, mock_identity): keystone_mixin = self.get_keystone_mixin() keystone_mixin.admin = mock.MagicMock() self.assertEqual(mock_identity.Identity.return_value, keystone_mixin._manager()) mock_identity.Identity.assert_called_once_with( keystone_mixin.admin) @mock.patch("%s.identity" % BASE) def test_delete(self, mock_identity): keystone_mixin = self.get_keystone_mixin() keystone_mixin._resource = "some_resource" keystone_mixin.id = lambda: "id_a" keystone_mixin.admin = mock.MagicMock() keystone_mixin.delete() mock_identity.Identity.assert_called_once_with(keystone_mixin.admin) identity_service = mock_identity.Identity.return_value identity_service.delete_some_resource.assert_called_once_with("id_a") @mock.patch("%s.identity" % BASE) def test_list(self, mock_identity): keystone_mixin = self.get_keystone_mixin() keystone_mixin._resource = "some_resource2" keystone_mixin.admin = mock.MagicMock() identity = mock_identity.Identity self.assertSequenceEqual( identity.return_value.list_some_resource2s.return_value, keystone_mixin.list()) identity.assert_called_once_with(keystone_mixin.admin) identity.return_value.list_some_resource2s.assert_called_once_with() class KeystoneEc2TestCase(test.TestCase): def test_user_id_property(self): user_client = mock.Mock() admin_client = mock.Mock() manager = resources.KeystoneEc2(user=user_client, admin=admin_client) self.assertEqual(user_client.keystone.auth_ref.user_id, manager.user_id) def test_list(self): user_client = mock.Mock() admin_client = mock.Mock() with mock.patch("%s.identity.Identity" % BASE, autospec=True) as p: identity = p.return_value manager = resources.KeystoneEc2(user=user_client, admin=admin_client) self.assertEqual(identity.list_ec2credentials.return_value, manager.list()) p.assert_called_once_with(user_client) identity.list_ec2credentials.assert_called_once_with( manager.user_id) def test_delete(self): user_client = mock.Mock() admin_client = mock.Mock() raw_resource = mock.Mock() with mock.patch("%s.identity.Identity" % BASE, autospec=True) as p: manager = resources.KeystoneEc2(user=user_client, admin=admin_client, resource=raw_resource) manager.delete() p.assert_called_once_with(user_client) p.return_value.delete_ec2credential.assert_called_once_with( manager.user_id, access=raw_resource.access) class SwiftMixinTestCase(test.TestCase): def get_swift_mixin(self): swift_mixin = resources.SwiftMixin() swift_mixin._service = "swift" return swift_mixin def test_manager(self): swift_mixin = self.get_swift_mixin() swift_mixin.user = mock.MagicMock() self.assertEqual(swift_mixin.user.swift.return_value, swift_mixin._manager()) def test_id(self): swift_mixin = self.get_swift_mixin() swift_mixin.raw_resource = mock.MagicMock() self.assertEqual(swift_mixin.raw_resource, swift_mixin.id()) def test_name(self): swift = self.get_swift_mixin() swift.raw_resource = ["name1", "name2"] self.assertEqual("name2", swift.name()) def test_delete(self): swift_mixin = self.get_swift_mixin() swift_mixin.user = mock.MagicMock() swift_mixin._resource = "some_resource" swift_mixin.raw_resource = mock.MagicMock() swift_mixin.delete() swift_mixin.user.swift().delete_some_resource.assert_called_once_with( *swift_mixin.raw_resource) class SwiftObjectTestCase(test.TestCase): @mock.patch("%s.SwiftMixin._manager" % BASE) def test_list(self, mock_swift_mixin__manager): containers = [mock.MagicMock(), mock.MagicMock()] objects = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock()] mock_swift_mixin__manager().get_account.return_value = ( "header", containers) mock_swift_mixin__manager().get_container.return_value = ( "header", objects) self.assertEqual(len(containers), len(resources.SwiftContainer().list())) self.assertEqual(len(containers) * len(objects), len(resources.SwiftObject().list())) class SwiftContainerTestCase(test.TestCase): @mock.patch("%s.SwiftMixin._manager" % BASE) def test_list(self, mock_swift_mixin__manager): containers = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock()] mock_swift_mixin__manager().get_account.return_value = ( "header", containers) self.assertEqual(len(containers), len(resources.SwiftContainer().list())) class ManilaShareTestCase(test.TestCase): def test_list(self): share_resource = resources.ManilaShare() share_resource._manager = mock.MagicMock() share_resource.list() self.assertEqual("shares", share_resource._resource) share_resource._manager.return_value.list.assert_called_once_with() def test_delete(self): share_resource = resources.ManilaShare() share_resource._manager = mock.MagicMock() share_resource.id = lambda: "fake_id" share_resource.delete() self.assertEqual("shares", share_resource._resource) share_resource._manager.return_value.delete.assert_called_once_with( "fake_id") class ManilaShareNetworkTestCase(test.TestCase): def test_list(self): sn_resource = resources.ManilaShareNetwork() sn_resource._manager = mock.MagicMock() sn_resource.list() self.assertEqual("share_networks", sn_resource._resource) sn_resource._manager.return_value.list.assert_called_once_with() def test_delete(self): sn_resource = resources.ManilaShareNetwork() sn_resource._manager = mock.MagicMock() sn_resource.id = lambda: "fake_id" sn_resource.delete() self.assertEqual("share_networks", sn_resource._resource) sn_resource._manager.return_value.delete.assert_called_once_with( "fake_id") class ManilaSecurityServiceTestCase(test.TestCase): def test_list(self): ss_resource = resources.ManilaSecurityService() ss_resource._manager = mock.MagicMock() ss_resource.list() self.assertEqual("security_services", ss_resource._resource) ss_resource._manager.return_value.list.assert_called_once_with() def test_delete(self): ss_resource = resources.ManilaSecurityService() ss_resource._manager = mock.MagicMock() ss_resource.id = lambda: "fake_id" ss_resource.delete() self.assertEqual("security_services", ss_resource._resource) ss_resource._manager.return_value.delete.assert_called_once_with( "fake_id") class MistralWorkbookTestCase(test.TestCase): def test_delete(self): clients = mock.MagicMock() resource = mock.Mock() resource.name = "TEST_NAME" mistral = resources.MistralWorkbooks( user=clients, resource=resource) mistral.delete() clients.mistral().workbooks.delete.assert_called_once_with( "TEST_NAME") class MistralExecutionsTestCase(test.TestCase): def test_name(self): execution = mock.MagicMock(workflow_name="bar") execution.name = "foo" self.assertEqual("bar", resources.MistralExecutions(execution).name()) class SenlinMixinTestCase(test.TestCase): def test_id(self): senlin = resources.SenlinMixin() senlin.raw_resource = {"id": "TEST_ID"} self.assertEqual("TEST_ID", senlin.id()) def test__manager(self): senlin = resources.SenlinMixin() senlin._service = "senlin" senlin.user = mock.MagicMock() self.assertEqual(senlin.user.senlin.return_value, senlin._manager()) def test_list(self): senlin = resources.SenlinMixin() senlin._service = "senlin" senlin.user = mock.MagicMock() senlin._resource = "some_resources" some_resources = [{"name": "resource1"}, {"name": "resource2"}] senlin.user.senlin().some_resources.return_value = some_resources self.assertEqual(some_resources, senlin.list()) senlin.user.senlin().some_resources.assert_called_once_with() def test_delete(self): senlin = resources.SenlinMixin() senlin._service = "senlin" senlin.user = mock.MagicMock() senlin._resource = "some_resources" senlin.raw_resource = {"id": "TEST_ID"} senlin.user.senlin().delete_some_resource.return_value = None senlin.delete() senlin.user.senlin().delete_some_resource.assert_called_once_with( "TEST_ID") class WatcherTemplateTestCase(test.TestCase): def test_id(self): watcher = resources.WatcherTemplate() watcher.raw_resource = mock.MagicMock(uuid=100) self.assertEqual(100, watcher.id()) @mock.patch("%s.WatcherTemplate._manager" % BASE) def test_is_deleted(self, mock__manager): mock__manager.return_value.get.return_value = None watcher = resources.WatcherTemplate() watcher.id = mock.Mock() self.assertFalse(watcher.is_deleted()) mock__manager.side_effect = [watcher_exceptions.NotFound()] self.assertTrue(watcher.is_deleted()) def test_list(self): watcher = resources.WatcherTemplate() watcher._manager = mock.MagicMock() watcher.list() self.assertEqual("audit_template", watcher._resource) watcher._manager().list.assert_called_once_with(limit=0) class WatcherAuditTestCase(test.TestCase): def test_id(self): watcher = resources.WatcherAudit() watcher.raw_resource = mock.MagicMock(uuid=100) self.assertEqual(100, watcher.id()) def test_name(self): watcher = resources.WatcherAudit() watcher.raw_resource = mock.MagicMock(uuid="name") self.assertEqual("name", watcher.name()) @mock.patch("%s.WatcherAudit._manager" % BASE) def test_is_deleted(self, mock__manager): mock__manager.return_value.get.return_value = None watcher = resources.WatcherAudit() watcher.id = mock.Mock() self.assertFalse(watcher.is_deleted()) mock__manager.side_effect = [watcher_exceptions.NotFound()] self.assertTrue(watcher.is_deleted()) def test_list(self): watcher = resources.WatcherAudit() watcher._manager = mock.MagicMock() watcher.list() self.assertEqual("audit", watcher._resource) watcher._manager().list.assert_called_once_with(limit=0) class WatcherActionPlanTestCase(test.TestCase): def test_id(self): watcher = resources.WatcherActionPlan() watcher.raw_resource = mock.MagicMock(uuid=100) self.assertEqual(100, watcher.id()) def test_name(self): watcher = resources.WatcherActionPlan() self.assertIsInstance(watcher.name(), resources.base.NoName) @mock.patch("%s.WatcherActionPlan._manager" % BASE) def test_is_deleted(self, mock__manager): mock__manager.return_value.get.return_value = None watcher = resources.WatcherActionPlan() watcher.id = mock.Mock() self.assertFalse(watcher.is_deleted()) mock__manager.side_effect = [watcher_exceptions.NotFound()] self.assertTrue(watcher.is_deleted()) def test_list(self): watcher = resources.WatcherActionPlan() watcher._manager = mock.MagicMock() watcher.list() self.assertEqual("action_plan", watcher._resource) watcher._manager().list.assert_called_once_with(limit=0) class CinderImageVolumeCacheTestCase(test.TestCase): class Resource(object): def __init__(self, id=None, name=None): self.id = id self.name = name @mock.patch("rally_openstack.services.image.image.Image") def test_list(self, mock_image): admin = mock.Mock() glance = mock_image.return_value cinder = admin.cinder.return_value image_1 = self.Resource("foo", name="foo-name") image_2 = self.Resource("bar", name="bar-name") glance.list_images.return_value = [image_1, image_2] volume_1 = self.Resource(name="v1") volume_2 = self.Resource(name="image-foo") volume_3 = self.Resource(name="foo") volume_4 = self.Resource(name="bar") cinder.volumes.list.return_value = [volume_1, volume_2, volume_3, volume_4] manager = resources.CinderImageVolumeCache(admin=admin) self.assertEqual([{"volume": volume_2, "image": image_1}], manager.list()) mock_image.assert_called_once_with(admin) glance.list_images.assert_called_once_with() cinder.volumes.list.assert_called_once_with( search_opts={"all_tenants": 1}) def test_id_and_name(self): res = resources.CinderImageVolumeCache( {"volume": self.Resource("volume-id", "volume-name"), "image": self.Resource("image-id", "image-name")}) self.assertEqual("volume-id", res.id()) self.assertEqual("image-name", res.name()) class GnocchiMixinTestCase(test.TestCase): def get_gnocchi(self): gnocchi = resources.GnocchiMixin() gnocchi._service = "gnocchi" return gnocchi def test_id(self): gnocchi = self.get_gnocchi() gnocchi.raw_resource = {"name": "test_name"} self.assertEqual("test_name", gnocchi.id()) def test_name(self): gnocchi = self.get_gnocchi() gnocchi.raw_resource = {"name": "test_name"} self.assertEqual("test_name", gnocchi.name()) class GnocchiMetricTestCase(test.TestCase): def get_gnocchi(self): gnocchi = resources.GnocchiMetric() gnocchi._service = "gnocchi" return gnocchi def test_id(self): gnocchi = self.get_gnocchi() gnocchi.raw_resource = {"id": "test_id"} self.assertEqual("test_id", gnocchi.id()) def test_list(self): gnocchi = self.get_gnocchi() gnocchi._manager = mock.MagicMock() metrics = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock()] gnocchi._manager.return_value.list.side_effect = ( metrics[:2], metrics[2:4], []) self.assertEqual(metrics, gnocchi.list()) self.assertEqual( [mock.call(marker=None), mock.call(marker=metrics[1]["id"]), mock.call(marker=metrics[3]["id"])], gnocchi._manager.return_value.list.call_args_list) class GnocchiResourceTestCase(test.TestCase): def get_gnocchi(self): gnocchi = resources.GnocchiResource() gnocchi._service = "gnocchi" return gnocchi def test_id(self): gnocchi = self.get_gnocchi() gnocchi.raw_resource = {"id": "test_id"} self.assertEqual("test_id", gnocchi.id()) def test_list(self): gnocchi = self.get_gnocchi() gnocchi._manager = mock.MagicMock() res = [mock.MagicMock(), mock.MagicMock(), mock.MagicMock(), mock.MagicMock()] gnocchi._manager.return_value.list.side_effect = ( res[:2], res[2:4], []) self.assertEqual(res, gnocchi.list()) self.assertEqual( [mock.call(marker=None), mock.call(marker=res[1]["id"]), mock.call(marker=res[3]["id"])], gnocchi._manager.return_value.list.call_args_list)