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_ssl_context_with_proxy_is_respected(self):
# Create an ssl context which should not allow the negotiation with
# the `self.remote_website_https`.
bad_ctx = ssl.create_default_context(cafile=CERT_SELFSIGNED_CA)
geocoder_dummy = DummyGeocoder(proxies={"https": self.proxy_url},
ssl_context=bad_ctx,
timeout=self.timeout)
self.assertEqual(0, len(self.proxy_server.requests))
with self.assertRaises(GeocoderServiceError) as cm:
geocoder_dummy.geocode(self.remote_website_https)
self.assertIn('SSL', str(cm.exception))
self.assertEqual(1, len(self.proxy_server.requests))
# Non-geopy errors must not be swallowed
self.mock_func.side_effect = ValueError
with self.assertRaises(ValueError):
rl(sentinel.arg)
self.assertEqual(1, self.mock_func.call_count)
self.mock_func.reset_mock()
# geopy errors must be swallowed and retried
self.mock_func.side_effect = GeocoderServiceError
self.assertEqual(sentinel.return_value, rl(sentinel.arg))
self.assertEqual(4, self.mock_func.call_count)
self.mock_func.reset_mock()
# Successful value must be returned
self.mock_func.side_effect = [
GeocoderServiceError, GeocoderServiceError, sentinel.good
]
self.assertEqual(sentinel.good, rl(sentinel.arg))
self.assertEqual(3, self.mock_func.call_count)
self.mock_func.reset_mock()
# When swallowing is disabled, the exception must be raised
rl.swallow_exceptions = False
self.mock_func.side_effect = GeocoderQuotaExceeded
with self.assertRaises(GeocoderQuotaExceeded):
rl(sentinel.arg)
self.assertEqual(4, self.mock_func.call_count)
self.mock_func.reset_mock()
def __call__(self, *args, **kwargs):
self._sleep_between()
for i, is_last_try in zip(count(), _is_last_gen(self.max_retries)):
try:
return self.func(*args, **kwargs)
except GeocoderServiceError:
if not is_last_try:
logger.warning(
'RateLimiter caught an error, retrying '
'(%s/%s tries). Called with (*%r, **%r).',
i, self.max_retries, args, kwargs, exc_info=True
)
self._sleep(self.error_wait_seconds)
continue
if self.swallow_exceptions:
logger.warning(
'RateLimiter swallowed an error after %r retries. '
'Called with (*%r, **%r).',
i, args, kwargs, exc_info=True
)
return self.return_value_on_exception
class GeocoderQueryError(GeocoderServiceError):
"""
Either geopy detected input that would cause a request to fail,
or a request was made and the remote geocoding service responded
that the request was bad.
"""
class GeocoderQuotaExceeded(GeocoderServiceError):
"""
The remote geocoding service refused to fulfill the request
because the client has used its quota.
"""
class GeocoderAuthenticationFailure(GeocoderServiceError):
"""
The remote geocoding service rejected the API key or account
credentials this geocoder was instantiated with.
"""
class GeocoderInsufficientPrivileges(GeocoderServiceError):
"""
The remote geocoding service refused to fulfill a request using the
account credentials given.
"""
class GeocoderTimedOut(GeocoderServiceError):
"""
The call to the geocoding service was aborted because no response
the documentation of each geocoder's ``__init__`` for more details.
"""
class GeocoderServiceError(GeopyError):
"""
There was an exception caused when calling the remote geocoding service,
and no more specific exception could be raised by geopy. When calling
geocoders' ``geocode`` or `reverse` methods, this is the most generic
exception that can be raised, and any non-geopy exception will be caught
and turned into this. The exception's message will be that of the
original exception.
"""
class GeocoderQueryError(GeocoderServiceError):
"""
Either geopy detected input that would cause a request to fail,
or a request was made and the remote geocoding service responded
that the request was bad.
"""
class GeocoderQuotaExceeded(GeocoderServiceError):
"""
The remote geocoding service refused to fulfill the request
because the client has used its quota.
"""
class GeocoderAuthenticationFailure(GeocoderServiceError):
"""
print("Couldn't understand your input. Defaulting to automatic setup.")
import geopy
from geopy import GoogleV3
geocoder = GoogleV3(scheme='https')
# Warm-up geocode
try:
geocoder.geocode("123 5th Avenue, New York, NY")
except:
isthisisheresopythondoesntyellatme = True
try:
geocoder.geocode("123 5th Avenue, New York, NY")
print("The geocoder can operate with HTTPS enabled on your OS. Saving these changes...")
config['GEOCODER']['scheme'] = 'https'
print("Changes saved.")
except geopy.exc.GeocoderServiceError:
print("Geopy probably can't run without HTTPS (or your internet went down). Trying HTTP as the scheme...")
geocoder = GoogleV3(scheme='http')
try:
geocoder.geocode("123 5th Avenue, New York, NY")
print("The geocoder can operate, but without HTTPS enabled on your OS. Saving these changes...")
config['GEOCODER']['scheme'] = 'http'
print("Changes saved.")
except geopy.exc.GeocoderServiceError:
print("You probably don't have an internet connection, as HTTPS and HTTP validation both failed.",
"Defaulting to HTTP as the geopy scheme...", sep="\n")
config['GEOCODER']['scheme'] = 'http'
print("Changes saved.")
def _raise_for_error(self, body):
err = body.get('status')
if err:
code = err['value']
message = err['message']
# http://www.geonames.org/export/webservice-exception.html
if message.startswith("user account not enabled to use"):
raise GeocoderInsufficientPrivileges(message)
if code == 10:
raise GeocoderAuthenticationFailure(message)
if code in (18, 19, 20):
raise GeocoderQuotaExceeded(message)
raise GeocoderServiceError(message)
"""
Parse a location name, latitude, and longitude from an JSON response.
"""
status_code = doc.get("statusCode", 200)
if status_code != 200:
err = doc.get("errorDetails", "")
if status_code == 401:
raise GeocoderAuthenticationFailure(err)
elif status_code == 403:
raise GeocoderInsufficientPrivileges(err)
elif status_code == 429:
raise GeocoderQuotaExceeded(err)
elif status_code == 503:
raise GeocoderUnavailable(err)
else:
raise GeocoderServiceError(err)
try:
resources = doc['Response']['View'][0]['Result']
except IndexError:
resources = None
if not resources:
return None
def parse_resource(resource):
"""
Parse each return object.
"""
stripchars = ", \n"
addr = resource['Location']['Address']
address = addr.get('Label', '').strip(stripchars)
def _check_status(status):
"""
Validates error statuses.
"""
if status == 0:
# When there are no results, just return.
return
if status == 1:
raise GeocoderServiceError(
'Internal server error.'
)
elif status == 2:
raise GeocoderQueryError(
'Invalid request.'
)
elif status == 3:
raise GeocoderAuthenticationFailure(
'Authentication failure.'
)
elif status == 4:
raise GeocoderQuotaExceeded(
'Quota validate failure.'
)
elif status == 5:
raise GeocoderQueryError(