Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@wsgi.expected_errors((400, 403, 404))
@validation.schema(image_metadata.update)
def update(self, req, image_id, id, body):
context = req.environ['nova.context']
meta = body['meta']
if id not in meta:
expl = _('Request body and URI mismatch')
raise exc.HTTPBadRequest(explanation=expl)
image = self._get_image(context, image_id)
image['properties'][id] = meta[id]
common.check_img_metadata_properties_quota(context,
image['properties'])
try:
self.image_api.update(context, image_id, image, data=None,
@wsgi.expected_errors(400)
@validation.schema(assisted_volume_snapshots.snapshots_create)
def create(self, req, body):
"""Creates a new snapshot."""
context = req.environ['nova.context']
context.can(avs_policies.POLICY_ROOT % 'create')
snapshot = body['snapshot']
create_info = snapshot['create_info']
volume_id = snapshot['volume_id']
try:
return self.compute_api.volume_snapshot_create(context, volume_id,
create_info)
except (exception.VolumeBDMNotFound,
exception.VolumeBDMIsMultiAttach,
exception.InvalidVolume) as error:
@wsgi.expected_errors((404, 409))
def delete(self, req, id):
"""Destroys a server."""
try:
self._delete(req.environ['nova.context'], req, id)
except exception.InstanceNotFound:
msg = _("Instance could not be found")
raise exc.HTTPNotFound(explanation=msg)
except (exception.InstanceIsLocked,
exception.AllocationDeleteFailed) as e:
raise exc.HTTPConflict(explanation=e.format_message())
except exception.InstanceInvalidState as state_error:
common.raise_http_conflict_for_instance_invalid_state(state_error,
'delete', id)
@wsgi.expected_errors(409)
@wsgi.response(200)
@validation.schema(schema.create)
def create(self, req, body):
"""Creates a new agent build."""
context = req.environ['nova.context']
context.can(agents_policies.BASE_POLICY_NAME)
agent = body['agent']
hypervisor = agent['hypervisor']
os = agent['os']
architecture = agent['architecture']
version = agent['version']
url = agent['url']
md5hash = agent['md5hash']
agent_obj = objects.Agent(context=context)
@wsgi.expected_errors((400, 404, 501))
@validation.schema(hosts.update)
def update(self, req, id, body):
"""Return booleanized version of body dict.
:param Request req: The request object (containing 'nova-context'
env var).
:param str id: The host name.
:param dict body: example format {'host': {'status': 'enable',
'maintenance_mode': 'enable'}}
:return: Same dict as body but 'enable' strings for 'status' and
'maintenance_mode' are converted into True, else False.
:rtype: dict
"""
def read_enabled(orig_val):
# Convert enable/disable str to a bool.
val = orig_val.strip().lower()
@wsgi.expected_errors((400, 404, 409))
@wsgi.response(202)
@wsgi.action('removeSecurityGroup')
def _removeSecurityGroup(self, req, id, body):
context = req.environ['nova.context']
context.can(sg_policies.BASE_POLICY_NAME)
group_name = self._parse(body, 'removeSecurityGroup')
try:
return self._invoke(self.security_group_api.remove_from_instance,
context, id, group_name)
except (exception.SecurityGroupNotFound,
exception.InstanceNotFound) as exp:
raise exc.HTTPNotFound(explanation=exp.format_message())
except exception.NoUniqueMatch as exp:
raise exc.HTTPConflict(explanation=exp.format_message())
@wsgi.expected_errors((400, 404, 409))
@wsgi.action('evacuate')
@validation.schema(evacuate.evacuate, "2.0", "2.13")
@validation.schema(evacuate.evacuate_v214, "2.14", "2.28")
@validation.schema(evacuate.evacuate_v2_29, "2.29", "2.67")
@validation.schema(evacuate.evacuate_v2_68, "2.68")
def _evacuate(self, req, id, body):
"""Permit admins to evacuate a server from a failed host
to a new one.
"""
context = req.environ["nova.context"]
instance = common.get_instance(self.compute_api, context, id)
context.can(evac_policies.BASE_POLICY_NAME,
target={'user_id': instance.user_id,
'project_id': instance.project_id})
evacuate_body = body["evacuate"]
@wsgi.expected_errors((404, 409, 501))
@wsgi.action('unrescue')
def _unrescue(self, req, id, body):
"""Unrescue an instance."""
context = req.environ["nova.context"]
context.can(rescue_policies.BASE_POLICY_NAME)
instance = common.get_instance(self.compute_api, context, id)
try:
self.compute_api.unrescue(context, instance)
except exception.InstanceUnknownCell as e:
raise exc.HTTPNotFound(explanation=e.format_message())
except exception.InstanceIsLocked as e:
raise exc.HTTPConflict(explanation=e.format_message())
except exception.InstanceInvalidState as state_error:
common.raise_http_conflict_for_instance_invalid_state(state_error,
'unrescue',
id)
@wsgi.expected_errors((404, 409))
@wsgi.action('shelveOffload')
def _shelve_offload(self, req, id, body):
"""Force removal of a shelved instance from the compute node."""
context = req.environ["nova.context"]
context.can(shelve_policies.POLICY_ROOT % 'shelve_offload')
instance = common.get_instance(self.compute_api, context, id)
try:
self.compute_api.shelve_offload(context, instance)
except exception.InstanceUnknownCell as e:
raise exc.HTTPNotFound(explanation=e.format_message())
except exception.InstanceIsLocked as e:
raise exc.HTTPConflict(explanation=e.format_message())
except exception.InstanceInvalidState as state_error:
common.raise_http_conflict_for_instance_invalid_state(state_error,
'shelveOffload',
@wsgi.expected_errors(())
@validation.query_schema(quota_sets.query_schema_275, '2.75')
@validation.query_schema(quota_sets.query_schema, '2.0', '2.74')
@wsgi.response(202)
def delete(self, req, id):
context = req.environ['nova.context']
context.can(qs_policies.POLICY_ROOT % 'delete', {'project_id': id})
params = urlparse.parse_qs(req.environ.get('QUERY_STRING', ''))
user_id = params.get('user_id', [None])[0]
if user_id:
objects.Quotas.destroy_all_by_project_and_user(
context, id, user_id)
else:
objects.Quotas.destroy_all_by_project(context, id)