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_raise(self):
with self.assertRaises(JsonRpcServerError):
raise ServerError()
def unlock_wallet(passphrase):
""" RPC method to unlock wallet.
Args:
passphrase (str): The passphrase to use to unlock the wallet.
"""
global wallet
if not wallet['locked']:
raise ServerError("Wallet is already unlocked or does not use a passphrase.")
load_wallet(wallet_path=wallet['path'],
data_provider=wallet['data_provider'],
passphrase=passphrase)
def check_unlocked():
""" Raises an error if the wallet is locked.
"""
if wallet['obj'] is None or wallet['locked']:
raise ServerError("Wallet is locked. Use the 'unlock' command with the passphrase as an arg.")
def server_error(error):
"""Handle server errors caught by flask."""
# Note, in the case of a syntax error, we get a TypeError, in which case we
# should just use 500 Internal Server Error
code = 500
if hasattr(error, 'code'):
code = error.code
# Do they want a JSON response? (Check the accept header.) If so, handle it
# by returning JSON-RPC. Otherwise, return the default werkzeug response.
if request.accept_mimetypes and \
request_wants_json(request.accept_mimetypes):
return flask_error_response(code, \
str(exceptions.ServerError(HTTP_STATUS_CODES[code])))
else:
return default_exceptions[code]().get_response()
def _handle_exception(e):
logger.debug("exception: %s" % str(e))
data = dict(type=e.__class__.__name__,
message=str(e))
raise ServerError(data=json.dumps(data))
if await self.chain.validate_job_invocation(job_address, job_signature):
logger.debug("dispatching request to service; job_address: %s", job_address)
response = client.request(config.PASSTHROUGH_ENDPOINT, method, **kwargs)
db = self.app["db"]
job_entry = db.get(job_address, {})
job_entry["job_signature"] = job_signature
job_entry["completed"] = True
logger.debug("saving job to db; job_address: %s; db entry: %s", job_address, job_entry)
db[job_address] = job_entry
self.app.loop.create_task(self.chain.complete_job(job_address, job_signature))
logger.debug("returning response to client; job_address: %s", job_address)
return response
else:
logger.error("job invocation failed to validate")
raise ServerError("job invocation failed to validate")