Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
body = self.simulate_get(self.messages_path, self.project_id,
query_string='include_claimed=true',
headers=headers)
listed = json.loads(body[0])
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertEqual(len(listed['messages']), len(claimed))
now = timeutils.utcnow() + datetime.timedelta(seconds=10)
timeutils_utcnow = 'marconi.openstack.common.timeutils.utcnow'
with mock.patch(timeutils_utcnow) as mock_utcnow:
mock_utcnow.return_value = now
body = self.simulate_get(claim_href, self.project_id)
claim = json.loads(body[0])
self.assertEqual(self.srmock.status, falcon.HTTP_200)
self.assertEqual(self.srmock.headers_dict['Content-Location'],
claim_href)
self.assertEqual(claim['ttl'], 100)
## NOTE(cpp-cabrera): verify that claim age is non-negative
self.assertThat(claim['age'], matchers.GreaterThan(-1))
# Try to delete the message without submitting a claim_id
self.simulate_delete(message_href, self.project_id)
self.assertEqual(self.srmock.status, falcon.HTTP_403)
# Delete the message and its associated claim
self.simulate_delete(message_href, self.project_id,
query_string=params)
self.assertEqual(self.srmock.status, falcon.HTTP_204)
# Try to get it from the wrong project
def test_on_get_return_correct_data(self):
self.mock_db.get_action.return_value = get_fake_action_0()
self.resource.on_get(self.mock_req, self.mock_req, fake_action_0['action_id'])
result = self.mock_req.body
self.assertEqual(result, get_fake_action_0())
self.assertEqual(self.mock_req.status, falcon.HTTP_200)
db = MySQLdb.connect(**settings.dbConfig)
cursor = db.cursor()
raw_json = req.stream.read()
data = json.loads(raw_json, encoding='utf-8')
q = """INSERT INTO tes (field1, field2) VALUES(%s,%s)"""
cursor.execute(q, (data['field1'], data['field2']))
db.commit()
cursor.close()
output = {
'status': "Data berhasil disimpan"
}
resp.status = falcon.HTTP_200
data_resp = json.dumps(output, encoding='utf-8')
resp.body = data_resp
db.close()
except Exception as e:
db.rollback()
resp.body = json.dumps({'error':str(e)})
resp.status = falcon.HTTP_500
return resp
def on_post(self, req, resp):
metadata = MetadataDecoder(ScaleBytes(req.media.get('result')))
resp.status = falcon.HTTP_200
resp.media = metadata.process()
def on_options(self, req, res, id):
"""Acknowledge the OPTIONS method.
"""
res.status = falcon.HTTP_200
to_token = tokens.get_token(str(req.get_param('to')))
value = float(req.get_param('value'))
if from_token is None or to_token is None:
message = {
'success': False,
'message': 'From token OR To token is not found.'
}
else:
value = tokens.exchange(from_token, to_token, value)
message = {
'success': True,
'value': value
}
resp.status = falcon.HTTP_200
resp.body = json.dumps(message)
def on_get(self, req, resp, revision_id):
try:
entries = db_api.validation_get_all_entries(revision_id,
val_name=None)
except errors.RevisionNotFound as e:
raise falcon.HTTPNotFound(description=e.format_message())
resp.status = falcon.HTTP_200
resp.body = self.view_builder.detail(entries)
with excutils.save_and_reraise_exception():
LOG.exception(e.format_message())
for document in documents:
if secrets_manager.SecretsManager.requires_encryption(document):
policy.conditional_authorize(
'deckhand:create_encrypted_documents', req.context)
break
documents = self._encrypt_secret_documents(documents)
created_documents = self._create_revision_documents(
bucket_name, documents)
resp.body = self.view_builder.list(created_documents)
resp.status = falcon.HTTP_200
def on_get(self, req, resp):
resp.status = falcon.HTTP_200
resp.body = json.dumps({'status': 'UP'})
def on_success(self, resp, data=None):
resp.status = falcon.HTTP_200
meta = OrderedDict()
meta['code'] = 200
meta['message'] = 'OK'
obj = OrderedDict()
obj['meta'] = meta
if data:
BaseResource.find_datetime_to_ctime(data)
obj['data'] = data
resp.body = BaseResource.to_json(obj)