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_head(self):
response = self.simulate_head('/v1.1/ping')
self.assertEqual(self.srmock.status, falcon.HTTP_204)
self.assertEqual(response, [])
def test_metadata_deserialization(self):
# Normal case
self.simulate_put(self.queue_path + '/metadata',
self.project_id,
body='{"timespace": "Shangri-la"}')
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Too long
max_queue_metadata = 64
doc_tmpl = '{{"Dragon Torc":"{0}"}}'
doc_tmpl_ws = '{{ "Dragon Torc" : "{0}" }}' # with whitespace
envelope_length = len(doc_tmpl.format(''))
for tmpl in doc_tmpl, doc_tmpl_ws:
gen = '0' * (max_queue_metadata - envelope_length + 1)
doc = tmpl.format(gen)
self.simulate_put(self.queue_path + '/metadata',
self.project_id,
body=doc)
self.assertEqual(self.srmock.status, falcon.HTTP_400)
now = timeutils.utcnow_ts()
now_dt = datetime.datetime.utcfromtimestamp(now)
ttl = self._conf.transport.default_subscription_ttl
expires = now_dt + datetime.timedelta(seconds=ttl)
api_version = req.path.split('/')[1]
sub = self._subscription_controller.get(queue_name,
subscription_id,
project=project_id)
self._notification.send_confirm_notification(queue_name,
sub,
self._conf,
project_id,
str(expires),
api_version,
True)
resp.status = falcon.HTTP_204
resp.location = req.path
except storage_errors.SubscriptionDoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(six.text_type(ex))
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(six.text_type(ex))
except Exception:
description = (_(u'Subscription %(subscription_id)s could not be'
' confirmed.') %
dict(subscription_id=subscription_id))
LOG.exception(description)
raise falcon.HTTPBadRequest(_('Unable to confirm subscription'),
description)
def on_delete(self, req, res, notification_method_id):
helpers.validate_authorization(req, self._default_authorized_roles)
self._delete_notification(req.project_id, notification_method_id)
res.status = falcon.HTTP_204
def do_delete_stream_definitions(self, req, res, id):
helpers.validate_authorization(req, self._default_authorized_roles)
tenant_id = helpers.get_tenant_id(req)
self._stream_definition_delete(tenant_id, id)
res.status = falcon.HTTP_204
except Exception:
description = _(u'Claim could not be created.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Serialize claimed messages, if any. This logic assumes
# the storage driver returned well-formed messages.
if len(resp_msgs) != 0:
resp_msgs = [wsgi_utils.format_message_v1(
msg, req.path.rpartition('/')[0], cid) for msg in resp_msgs]
resp.location = req.path + '/' + cid
resp.body = utils.to_json(resp_msgs)
resp.status = falcon.HTTP_201
else:
resp.status = falcon.HTTP_204
def do_delete_transforms(self, req, res, transform_id):
helpers.validate_authorization(req, self._default_authorized_roles)
tenant_id = helpers.get_tenant_id(req)
self._delete_transform(tenant_id, transform_id)
res.status = falcon.HTTP_204
value=int(req.content_length))
tenant_id = (req.cross_project_id if req.cross_project_id
else req.project_id)
try:
self._processor.send_message(
logs=log_list,
global_dimensions=global_dimensions,
log_tenant_id=tenant_id
)
except Exception as ex:
res.status = getattr(ex, 'status', falcon.HTTP_500)
return
res.status = falcon.HTTP_204
def do_post_events(self, req, res):
helpers.validate_json_content_type(req)
helpers.validate_authorization(req, self._post_events_authorized_roles)
event = helpers.read_http_resource(req)
self._validate_event(event)
tenant_id = helpers.get_tenant_id(req)
transformed_event = self._event_transform(event, tenant_id,
self._region)
self._send_event(transformed_event)
res.status = falcon.HTTP_204
def on_get(self, req, resp):
"""
It really does nothing right now. It may do more later
"""
resp.status = falcon.HTTP_204