def test_APIGetWeirdError(self):
"""
Make sure we handle weird exceptions properly
We should see a bunch of retries
:return:
"""
import requests
# MonkeyPatching!!!
def fakefunc(a, headers=False):
raise Exception("I AM FAKE")
requests.__dict__['get'] = fakefunc
with self.assertRaises(NetworkException) as e:
result = APIGet('IAmaTeapot')
# Make sure we've freaked out appropriately
self.assertTrue(e.exception.message.index("Connection Exception:") == 0)
# Make sure we've done the requisite number of retries
self.assertEqual(int(e.exception.message[-1]), RETRIES_ALLOWED)
python类__dict__()的实例源码
def __init__(self, url_scheme=DEFAULT_SCHEME):
self.session = Session()
requests = self._get_global_requests_module()
# Methods to replace
self.methods = ('request', 'get', 'head', 'post',
'patch', 'put', 'delete', 'options')
# Store the original methods
self.orig_methods = dict(
(m, requests.__dict__[m]) for m in self.methods)
# Monkey patch
g = globals()
for m in self.methods:
requests.__dict__[m] = g[m]
def __exit__(self, *args):
requests = self._get_global_requests_module()
for m in self.methods:
requests.__dict__[m] = self.orig_methods[m]
# These are the same methods defined for the global requests object
def __api_request(self,url,method,**karg):
try:return requests.__dict__[method.lower().strip()](url,**karg)
except requests.exceptions.MissingSchema as e:
return self.__api_request("http://"+url,method,**karg)
except requests.exceptions.ConnectionError as e:
raise RuntimeError("Couldn't connect to server (unreachable). Check your network")
except KeyError as e:
raise RuntimeError("Invalid method name '%s' in api.json" % method)
def to_json(self, indent=None):
return json.dumps(self,
default=lambda o: ((float(o) if o % 1 > 0 else int(o))
if isinstance(o, decimal.Decimal) else o.__dict__),
sort_keys=True, indent=indent)
def apply_json(self, j):
if isinstance(j, str):
j = json.loads(j)
self.__dict__.update(j)
def __getattr__(self, name):
method = requests.__dict__.get(name.lower())
if not method:
return method
def _wrapper(*args, **kwargs):
if 'auth' not in kwargs:
kwargs['auth'] = NetrcBypassAuth()
if not self.verify_ssl and args[0].startswith('https://') and 'verify' not in kwargs:
kwargs['verify'] = False
return method(*args, **kwargs)
return _wrapper
# create class-of-a-class
def make_http_request(url, data=None, headers=None, method='GET'):
if is_string(method):
method = requests.__dict__[method.lower()]
return method(url, headers=headers, data=data, auth=NetrcBypassAuth(), verify=False)
def __getattr__(self, name):
method = requests.__dict__.get(name.lower())
if not method:
return method
def _missing(*args, **kwargs):
if 'auth' not in kwargs:
kwargs['auth'] = NetrcBypassAuth()
if 'verify' not in kwargs:
kwargs['verify'] = False
return method(*args, **kwargs)
return _missing
# create class-of-a-class
def make_http_request(url, data=None, headers=None, method='GET'):
if is_string(method):
method = requests.__dict__[method.lower()]
return method(url, headers=headers, data=data, auth=NetrcBypassAuth(), verify=False)
