1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
"""Context manager based WSGI interception.
"""
from importlib import import_module
from uuid import uuid4
from six.moves.urllib import parse as urlparse
import wsgi_intercept
class Interceptor(object):
"""A convenience class over the guts of wsgi_intercept.
An Interceptor subclass provides a clean entry point to the wsgi_intercept
functionality in two ways: by encapsulating the interception addition and
removal in methods and by providing a context manager that automates the
process of addition and removal.
Each Interceptor subclass is associated with a specific http library.
Each class may be passed a url or a host and a port. If no args are passed
a hostname will be automatically generated and the resulting url will be
returned by the context manager.
"""
def __init__(self, app, host=None, port=80, prefix=None, url=None):
assert app
if (not host and not url):
host = str(uuid4())
self.app = app
if url:
self._init_from_url(url)
self.url = url
else:
self.host = host
self.port = int(port)
self.script_name = prefix or ''
self.url = self._url_from_primitives()
self._module = import_module('.%s' % self.MODULE_NAME,
package='wsgi_intercept')
def __enter__(self):
self.install_intercept()
return self.url
def __exit__(self, exc_type, value, traceback):
self.uninstall_intercept()
def _url_from_primitives(self):
if self.port == 443:
scheme = 'https'
else:
scheme = 'http'
if self.port and self.port not in [443, 80]:
port = ':%s' % self.port
else:
port = ''
netloc = self.host + port
return urlparse.urlunsplit((scheme, netloc, self.script_name,
None, None))
def _init_from_url(self, url):
parsed_url = urlparse.urlsplit(url)
host, port = parsed_url.netloc.split(':')
if not port:
if parsed_url.scheme == 'https':
port = 443
else:
port = 80
path = parsed_url.path
if path == '/' or not path:
self.script_name = ''
else:
self.script_name = path
self.host = host
self.port = int(port)
def install_intercept(self):
self._module.install()
wsgi_intercept.add_wsgi_intercept(self.host, self.port, self.app,
script_name=self.script_name)
def uninstall_intercept(self):
wsgi_intercept.remove_wsgi_intercept(self.host, self.port)
self._module.uninstall()
class HttpClientInterceptor(Interceptor):
"""Interceptor for httplib and http.client."""
MODULE_NAME = 'http_client_intercept'
class Httplib2Interceptor(Interceptor):
"""Interceptor for httplib2."""
MODULE_NAME = 'httplib2_intercept'
class RequestsInterceptor(Interceptor):
"""Interceptor for requests."""
MODULE_NAME = 'requests_intercept'
class UrllibInterceptor(Interceptor):
"""Interceptor for urllib2 and urllib.request."""
MODULE_NAME = 'urllib_intercept'
|
