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_raises_error_if_server_is_unreachable(self, mocked_post):
mocked_post.side_effect = requests_exceptions.RequestException
with self.assertRaises(fxa.OAuth2Error):
fxa.trade_code(oauth_uri='https://unknown',
client_id='abc',
client_secret='cake',
code='1234')
def search(self, query):
if self._response:
return self._response
raise RequestException()
def get_version(self):
log.info('querying %s', self.software)
url = 'http://{host}:{port}/version'.format(host=self.host, port=self.port)
log.debug('GET %s', url)
try:
req = requests.get(url)
except requests.exceptions.RequestException as _:
qquit('CRITICAL', _)
log.debug("response: %s %s", req.status_code, req.reason)
log.debug("content:\n%s\n%s\n%s", '='*80, req.content.strip(), '='*80)
# Special handling for Nginx, expecting 404 rather than usual 200
if req.status_code != 404:
qquit('CRITICAL', '{0} {1} (expecting 404)'.format(req.status_code, req.reason))
soup = BeautifulSoup(req.content, 'html.parser')
if log.isEnabledFor(logging.DEBUG):
log.debug("BeautifulSoup prettified:\n{0}\n{1}".format(soup.prettify(), '='*80))
try:
version = soup.findAll('center')[1].text
except (AttributeError, TypeError) as _:
qquit('UNKNOWN', 'failed to find parse {0} output. {1}\n{2}'\
.format(self.software, support_msg_api(), traceback.format_exc()))
if '/' in version:
version = version.split('/')[1]
def _perform_request(method, path, **kw):
url = "https://" + GLOBUS_AUTH_HOST + path
try:
# Raises ConnectionError
r = method(url, timeout=15, **kw)
# Raises HTTPError
r.raise_for_status()
except requests.exceptions.RequestException as e:
raise GlobusAuthError(e.message)
return r
for msg in messages:
if isinstance(msg, str):
assert not isinstance(messages, str) # because we would split a string to pieces.
msg = TextMessage(msg, parse_mode="text")
# end if
if not isinstance(msg, (Message, SendableMessageBase)):
raise TypeError("Is not a Message/SendableMessageBase type.")
# end if
# if msg._next_msg: # TODO: Reply message?
# message.insert(message.index(msg) + 1, msg._next_msg)
# msg._next_msg = None
from requests.exceptions import RequestException
msg._apply_update_receiver(receiver=reply_chat, reply_id=reply_msg)
try:
yield msg.send(self.bot)
except (TgApiException, RequestException):
logger.exception("Manager failed messages. Message was {msg!s}".format(msg=msg))
# end try
def new_albums(self, offset=0, limit=50):
action = 'http://music.163.com/api/album/new?area=ALL&offset={}&total=true&limit={}'.format( # NOQA
offset, limit)
try:
data = self.httpRequest('GET', action)
return data['albums']
except requests.exceptions.RequestException as e:
log.error(e)
return []
ec == const.IETaskNotFound or # 36016 or # sc == 404 Task was not found
# the following was found by xslidian, but i have never ecountered before
ec == 31390): # sc == 404 # {"error_code":31390,"error_msg":"Illegal File"} # r.url.find('http://bcscdn.baidu.com/bcs-cdn/wenxintishi') == 0
result = ec
# TODO: Move this out to cdl_cancel() ?
#if ec == const.IETaskNotFound:
# pr(r.json())
if dumpex:
self.__dump_exception(None, url, pars, r, act)
else:
# gate for child classes to customize behaviors
# the function should return ERequestFailed if it doesn't handle the case
result = self.__handle_more_response_error(r, sc, ec, act, actargs)
if result == const.ERequestFailed and dumpex:
self.__dump_exception(None, url, pars, r, act)
except (requests.exceptions.RequestException,
socket.error,
ReadTimeoutError) as ex:
# If certificate check failed, no need to continue
# but prompt the user for work-around and quit
# why so kludge? because requests' SSLError doesn't set
# the errno and strerror due to using **kwargs,
# so we are forced to use string matching
if isinstance(ex, requests.exceptions.SSLError) \
and re.match(r'^\[Errno 1\].*error:14090086.*:certificate verify failed$', str(ex), re.I):
# [Errno 1] _ssl.c:504: error:14090086:SSL routines:SSL3_GET_SERVER_CERTIFICATE:certificate verify failed
result = const.EFatal
self.__dump_exception(ex, url, pars, r, act)
perr("\n\n== Baidu's Certificate Verification Failure ==\n"
"We couldn't verify Baidu's SSL Certificate.\n"
"It's most likely that the system doesn't have "
"the corresponding CA certificate installed.\n"
fail_with_message(event, failure_message)
notify_bot_owner(event, request_data, response.status_code, response.content)
except requests.exceptions.Timeout:
logging.info("Trigger event %s on %s timed out. Retrying" % (
event["command"], event['service_name']))
failure_message = "A timeout occurred."
request_retry(event, request_data, failure_message=failure_message)
except requests.exceptions.ConnectionError:
logging.info("Trigger event %s on %s resulted in a connection error. Retrying"
% (event["command"], event['service_name']))
failure_message = "A connection error occurred. Is my bot server down?"
request_retry(event, request_data, failure_message=failure_message)
except requests.exceptions.RequestException as e:
response_message = ("An exception of type *%s* occurred for message `%s`! "
"See the Zulip server logs for more information." % (
type(e).__name__, event["command"],))
logging.exception("Outhook trigger failed:\n %s" % (e,))
fail_with_message(event, response_message)
notify_bot_owner(event, request_data, exception=e)
def _get_query_monitoring_data(self, instance):
query_data = None
query_monitoring_url = instance.get('query_monitoring_url')
if query_monitoring_url:
url = '{}{}'.format(query_monitoring_url, COUCHBASE_VITALS_PATH)
try:
query_data = self._get_stats(url)
except requests.exceptions.RequestException:
self.log.error(
"Error accessing the endpoint %s, make sure you're running at least "
"couchbase 4.5 to collect the query monitoring metrics",
url,
)
return query_data
mt.addUIMessage("No SSL certs were found on this ip: {ip}".format(ip=ip))
mt.returnOutput()
else:
if request.status_code == 400:
results = request.json()
mt.addException(str(results['error']))
if request.status_code == 429:
results = request.json()
mt.addException(str(results['error']))
if request.status_code == 404:
mt.addException("No SSL certs were found on this ip: {ip}".format(ip=ip))
if request.status_code == 500:
mt.addException("There has been a server error!!!")
mt.throwExceptions()
except requests.exceptions.RequestException as e:
mt.addException(str(e))
mt.throwExceptions()