Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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 mock_create(container_config, *args, **kwargs):
"""Mocks the container create call, returns the mock container object
and also ensures that the container_config is correct
"""
config = container_config['config']
# Values below should not be driven by the values in container_options
assert config['security.privileged'] != 'invalid'
assert config['user.lxdock.homedir'] != 'invalid'
assert config['user.lxdock.made'] != 'invalid'
# Value below is a custom value that should be passed from container_options
assert config['valid_key'] == 'valid_value'
return cont_return
client_mock = unittest.mock.Mock(**{
'containers.create.side_effect': mock_create,
'containers.get.side_effect': NotFound(''),
})
container = Container('myproject', THIS_DIR, client_mock, **container_options)
assert container._get_container() is cont_return
assert client_mock.containers.get.called
assert client_mock.containers.create.called
#
# 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 = {}
if mode is not None:
if isinstance(mode, int):
mode = oct(mode)
elif not mode.startswith('0'):
mode = '0{0}'.format(mode)
headers['X-LXD-mode'] = mode
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:
The model lifecycle is this: A model's get/create methods will
return an instance. That instance may or may not be a partial
instance. If it is a partial instance, `sync` will be called
and the rest of the object retrieved from the server when
un-initialized attributes are read. When attributes are modified,
the instance is marked as dirty. `save` will save the changes
to the server.
If the LXD server sends attributes that this version of pylxd is unaware of
then a warning is printed. By default the warning is issued ONCE and then
supressed for every subsequent attempted setting. The warnings can be
completely suppressed by setting the environment variable PYLXD_WARNINGS to
'none', or always displayed by setting the PYLXD_WARNINGS variable to
'always'.
"""
NotFound = exceptions.NotFound
__slots__ = ['client', '__dirty__']
def __init__(self, client, **kwargs):
self.__dirty__ = set()
self.client = client
for key, val in kwargs.items():
try:
setattr(self, key, val)
except AttributeError:
global _seen_attribute_warnings
env = os.environ.get('PYLXD_WARNINGS', '').lower()
item = "{}.{}".format(self.__class__.__name__, key)
if env != 'always' and item in _seen_attribute_warnings:
continue
_seen_attribute_warnings.add(item)
def _get_container(self, create=True):
""" Gets or creates the PyLXD container. """
try:
container = self.client.containers.get(self.lxd_name)
except NotFound:
container = None
else:
return container
if not create:
return
logger.warning('Unable to find container "{name}" for directory "{homedir}"'.format(
name=self.name, homedir=self.homedir))
logger.info(
'Creating new container "{name}" '
'from image {image}'.format(name=self.lxd_name, image=self.options['image']))
privileged = self.options.get('privileged', False)
mode = self.options.get('mode', 'pull')