It checks all openstack resources before cleaning.
It allows removing images even if the test fail quickly.
Change-Id: I1763b8a70fa7ffeb03c435e8e775c7afd1a7adac
Signed-off-by: Cédric Ollivier <cedric.ollivier@orange.com>
assert self.orig_cloud
assert self.cloud
super(VmReady1, self).clean()
- self.cloud.delete_image(self.image.id)
- self.orig_cloud.delete_flavor(self.flavor.id)
+ if self.image:
+ self.cloud.delete_image(self.image.id)
+ if self.flavor:
+ self.orig_cloud.delete_flavor(self.flavor.id)
except Exception: # pylint: disable=broad-except
self.__logger.exception("Cannot clean all ressources")
try:
assert self.orig_cloud
assert self.cloud
- self.cloud.delete_floating_ip(self.fip.id)
- self.cloud.delete_server(self.sshvm, wait=True)
- self.cloud.delete_security_group(self.sec.id)
- self.cloud.delete_keypair(self.keypair.name)
+ if self.fip:
+ self.cloud.delete_floating_ip(self.fip.id)
+ if self.sshvm:
+ self.cloud.delete_server(self.sshvm, wait=True)
+ if self.sec:
+ self.cloud.delete_security_group(self.sec.id)
+ if self.keypair:
+ self.cloud.delete_keypair(self.keypair.name)
super(SingleVm1, self).clean()
except Exception: # pylint: disable=broad-except
self.__logger.exception("Cannot clean all ressources")
"""Remove projects/users"""
try:
assert self.orig_cloud
- assert self.user.id
- assert self.project.id
- self.orig_cloud.delete_user(self.user.id)
- self.orig_cloud.delete_project(self.project.id)
+ if self.user:
+ self.orig_cloud.delete_user(self.user.id)
+ if self.project:
+ self.orig_cloud.delete_project(self.project.id)
if self.role:
self.orig_cloud.delete_role(self.role.id)
except Exception: # pylint: disable=broad-except
def clean(self):
try:
assert self.cloud
- self.cloud.remove_router_interface(self.router, self.subnet.id)
- self.cloud.delete_router(self.router.id)
- self.cloud.delete_subnet(self.subnet.id)
- self.cloud.delete_network(self.network.id)
+ if self.router:
+ if self.subnet:
+ self.cloud.remove_router_interface(
+ self.router, self.subnet.id)
+ self.cloud.delete_router(self.router.id)
+ if self.subnet:
+ self.cloud.delete_subnet(self.subnet.id)
+ if self.network:
+ self.cloud.delete_network(self.network.id)
except Exception: # pylint: disable=broad-except
self.__logger.exception("cannot clean all ressources")
def clean(self):
assert self.cloud
- self.cloud.delete_server(
- self.vm2, wait=True,
- timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
- self.cloud.delete_floating_ip(self.fip2.id)
- self.cloud.delete_volume(self.volume.id)
+ if self.vm2:
+ self.cloud.delete_server(
+ self.vm2, wait=True,
+ timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
+ if self.fip2:
+ self.cloud.delete_floating_ip(self.fip2.id)
+ if self.volume:
+ self.cloud.delete_volume(self.volume.id)
super(CinderCheck, self).clean()
def clean(self):
"""Cleanup of OpenStack resources. Should be called on completion."""
super(RallyBase, self).clean()
- self.orig_cloud.delete_flavor(self.flavor_alt.id)
+ if self.flavor_alt:
+ self.orig_cloud.delete_flavor(self.flavor_alt.id)
def is_successful(self):
"""The overall result of the test."""
Cleanup all OpenStack objects. Should be called on completion.
"""
super(TempestCommon, self).clean()
- self.cloud.delete_image(self.image_alt)
- self.orig_cloud.delete_flavor(self.flavor_alt.id)
+ if self.image_alt:
+ self.cloud.delete_image(self.image_alt)
+ if self.flavor_alt:
+ self.orig_cloud.delete_flavor(self.flavor_alt.id)
assert self.cloud
os.remove(self.privkey_filename)
os.remove(self.pubkey_filename)
- self.cloud.delete_image(self.image)
+ if self.image:
+ self.cloud.delete_image(self.image)
self.cloud.delete_network("pns-internal-net_{}".format(self.guid))
self.cloud.delete_network("pns-internal-net2_{}".format(self.guid))
except Exception: # pylint: disable=broad-except
def clean(self):
assert self.cloud
- self.cloud.delete_server(
- self.vm2, wait=True,
- timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
+ if self.vm2:
+ self.cloud.delete_server(
+ self.vm2, wait=True,
+ timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
super(VPingSSH, self).clean()
def clean(self):
assert self.cloud
- self.cloud.delete_server(
- self.vm1, wait=True,
- timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
- self.cloud.delete_server(
- self.vm2, wait=True,
- timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
+ if self.vm1:
+ self.cloud.delete_server(
+ self.vm1, wait=True,
+ timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
+ if self.vm2:
+ self.cloud.delete_server(
+ self.vm2, wait=True,
+ timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
super(VPingUserdata, self).clean()
self.vping.clean()
def test_clean_exc2(self):
+ self.vping.vm2 = munch.Munch(id='vm2')
mdelete_server = self.vping.cloud.delete_server
mdelete_server.side_effect = shade.OpenStackCloudException(None)
with self.assertRaises(shade.OpenStackCloudException):
@mock.patch('functest.core.singlevm.SingleVm2.clean',
side_effect=Exception)
def test_clean_exc3(self, *args):
- self.vping.vm2 = munch.Munch()
+ self.vping.vm2 = munch.Munch(id='vm2')
with self.assertRaises(Exception):
self.vping.clean()
self.vping.cloud.delete_server.assert_called_once_with(
args[0].assert_called_once_with()
@mock.patch('functest.core.singlevm.SingleVm2.clean')
- def test_clean(self, *args):
- self.vping.vm2 = munch.Munch()
+ def test_clean1(self, *args):
+ self.vping.vm2 = None
+ self.vping.clean()
+ self.vping.cloud.delete_server.assert_not_called()
+ args[0].assert_called_once_with()
+
+ @mock.patch('functest.core.singlevm.SingleVm2.clean')
+ def test_clean2(self, *args):
+ self.vping.vm2 = munch.Munch(id='vm2')
self.vping.clean()
self.vping.cloud.delete_server.assert_called_once_with(
self.vping.vm2, wait=True,