Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_delete(self):
"""A profile is deleted."""
self.profile.delete()
self.assertRaises(
exceptions.LXDAPIException,
self.client.profiles.get, self.profile.name)
def delete_profile(self, name):
"""Delete a profile."""
try:
self.lxd.profiles[name].delete()
except exceptions.LXDAPIException as e:
if e.response.status_code == 404:
return
raise
def shutdown(self, wait=True, retry=1):
"""Shutdown instance."""
if self.pylxd_container.status == 'Stopped':
return
try:
LOG.debug("%s: shutting down (wait=%s)", self, wait)
self.pylxd_container.stop(wait=wait)
except (pylxd_exc.LXDAPIException, pylxd_exc.NotFound) as e:
# An exception happens here sometimes (LP: #1783198)
# LOG it, and try again.
LOG.warning(
("%s: shutdown(retry=%d) caught %s in shutdown "
"(response=%s): %s"),
self, retry, e.__class__.__name__, e.response, e)
if isinstance(e, pylxd_exc.NotFound):
LOG.debug("container_exists(%s) == %s",
self.name, self.platform.container_exists(self.name))
if retry == 0:
raise e
return self.shutdown(wait=wait, retry=retry - 1)
def wait(self):
"""Wait for the operation to complete and return."""
response = self._client.api.operations[self.id].wait.get()
try:
if response.json()['metadata']['status'] == 'Failure':
raise exceptions.LXDAPIException(response)
except KeyError:
# Support for legacy LXD
pass
if src_verify_cert is None:
src_verify_cert = verify_cert
container = container_get(
name, src_remote_addr, src_cert, src_key, src_verify_cert, _raw=True
)
dest_client = pylxd_client_get(
remote_addr, cert, key, verify_cert
)
for pname in container.profiles:
try:
dest_client.profiles.get(pname)
except pylxd.exceptions.LXDAPIException:
raise SaltInvocationError(
'not all the profiles from the source exist on the target'
)
was_running = container.status_code == CONTAINER_STATUS_RUNNING
if stop_and_start and was_running:
container.stop(wait=True)
try:
dest_container = container.migrate(dest_client, wait=True)
dest_container.profiles = container.profiles
dest_container.save()
except pylxd.exceptions.LXDAPIException as e:
raise CommandExecutionError(six.text_type(e))
# Remove the source container
def halt():
config = get_config()
container = get_container()
if container.status_code == ContainerStatus.Stopped:
print("The container is already stopped.")
return
hostnames = config.get('hostnames', [])
if hostnames:
_unsetup_hostnames(container, hostnames)
print("Stopping...")
try:
container.stop(timeout=30, force=False, wait=True)
except pylxd.exceptions.LXDAPIException:
print("Can't stop the container. Forcing...")
container.stop(force=True, wait=True)
'''Translates a plyxd model object to a dict'''
marshalled = {}
for key in obj.__attributes__.keys():
if hasattr(obj, key):
marshalled[key] = getattr(obj, key)
return marshalled
#
# Monkey patching for missing functionality in pylxd
#
if HAS_PYLXD:
import pylxd.exceptions # NOQA
if not hasattr(pylxd.exceptions, 'NotFound'):
# Old version of pylxd
class NotFound(pylxd.exceptions.LXDAPIException):
'''An exception raised when an object is not found.'''
pylxd.exceptions.NotFound = NotFound
try:
from pylxd.container import Container
except ImportError:
from pylxd.models.container import Container
class FilesManager(Container.FilesManager):
def put(self, filepath, data, mode=None, uid=None, gid=None):
headers = {}