Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def delete_container(self, name, enforce=False):
"""Delete a container in lxd."""
# enforce is a hack. There's a race somewhere in the delete.
# To ensure we don't get an infinite loop, let's count.
count = 0
try:
result = self.lxd['containers'][name].delete()
except exceptions.LXDAPIException as e:
if e.response.status_code in (400, 404):
return
raise
while enforce and result.status_code == 404 and count < 10:
try:
result = self.lxd['containers'][name].delete()
except exceptions.LXDAPIException as e:
if e.response.status_code in (400, 404):
return
raise
count += 1
try:
operation_uuid = result.json()['operation'].split('/')[-1]
result = self.lxd.operations[operation_uuid].wait.get()
except KeyError:
pass # 404 cases are okay.
def test_delete(self):
"""The container is deleted."""
self.container.delete(wait=True)
self.assertRaises(
exceptions.LXDAPIException,
self.client.containers.get, self.container.name)
def test_delete(self):
"""A network is deleted"""
self.network.delete()
self.assertRaises(
exceptions.LXDAPIException,
self.client.networks.get, self.network.name)
return
try:
data = response.json()
except ValueError:
# Not a JSON response
return
if response.status_code == 200:
# Synchronous request
try:
if data['type'] != 'sync':
raise exceptions.LXDAPIException(response)
except KeyError:
# Missing 'type' in response
raise exceptions.LXDAPIException(response)
def _assert_response(self, response, allowed_status_codes=(200,),
stream=False, is_api=True):
"""Assert properties of the response.
LXD's API clearly defines specific responses. If the API
response is something unexpected (i.e. an error), then
we need to raise an exception and have the call points
handle the errors or just let the issue be raised to the
user.
"""
if response.status_code not in allowed_status_codes:
if response.status_code == 404:
raise exceptions.NotFound(response)
raise exceptions.LXDAPIException(response)
# In the case of streaming, we can't validate the json the way we
# would with normal HTTP responses, so just ignore that entirely.
# Likewize, we can ignore NON api calls as the contents don't need to
# be validated.
if stream or not is_api:
return
try:
data = response.json()
except ValueError:
# Not a JSON response
return
if response.status_code == 200:
# Synchronous request
def __init__(self, response):
super(LXDAPIException, self).__init__()
self.response = response
# Likewize, we can ignore NON api calls as the contents don't need to
# be validated.
if stream or not is_api:
return
try:
data = response.json()
except ValueError:
# Not a JSON response
return
if response.status_code == 200:
# Synchronous request
try:
if data['type'] != 'sync':
raise exceptions.LXDAPIException(response)
except KeyError:
# Missing 'type' in response
raise exceptions.LXDAPIException(response)