Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _on_request(self, _request, conn, method, url,
body=None, headers=None, **kw):
# Create request contract based on incoming params
req = Request(method)
req.headers = headers or {}
req.body = body
# Compose URL
req.url = 'http://{}:{}{}'.format(conn.host, conn.port, url)
# Match the request against the registered mocks in pook
mock = self.engine.match(req)
# If cannot match any mock, run real HTTP request since networking,
# otherwise this statement won't be reached
# (an exception will be raised before).
if not mock:
return _request(conn, method, url,
body=body, headers=headers, **kw)
def _on_request(self, urlopen, path, pool, method, url,
body=None, headers=None, **kw):
# Remove bypass headers
real_headers = dict(headers or {})
real_headers.pop(URLLIB3_BYPASS)
# Create request contract based on incoming params
req = Request(method)
req.headers = real_headers
req.body = body
# Compose URL
req.url = '{}://{}:{:d}{}'.format(
pool.scheme,
pool.host,
pool.port or 80,
url
)
# Match the request against the registered mocks in pook
mock = self.engine.match(req)
# If cannot match any mock, run real HTTP request since networking
# or silent model will be enabled, otherwise this statement won't
def _on_request(self, _request, session, method, url,
data=None, headers=None, **kw):
# Create request contract based on incoming params
req = Request(method)
req.headers = headers or {}
req.body = data
# Expose extra variadic arguments
req.extra = kw
# Compose URL
req.url = str(url)
# Match the request against the registered mocks in pook
mock = self.engine.match(req)
# If cannot match any mock, run real HTTP request if networking
# or silent model are enabled, otherwise this statement won't
# be reached (an exception will be raised before).
if not mock:
# Stores the number of times the mock should live
self._times = 1
# Stores the number of times the mock has been matched
self._matches = 0
# Stores the simulated error exception
self._error = None
# Stores the optional network delay in milliseconds
self._delay = 0
# Stores the mock persistance mode. `True` means it will live forever
self._persist = False
# Optional binded engine where the mock belongs to
self._engine = None
# Store request-response mock matched calls
self._calls = []
# Stores the input request instance
self._request = request or Request()
# Stores the response mock instance
self._response = response or Response()
# Stores the mock matcher engine used for outgoing traffic matching
self.matchers = MatcherEngine()
# Stores filters used to filter outgoing HTTP requests.
self.filters = []
# Stores HTTP request mappers used by the mock.
self.mappers = []
# Stores callback functions that will be triggered if the mock
# matches outgoing traffic.
self.callbacks = []
# Triggers instance methods based on argument names
trigger_methods(self, kw)
# Trigger matchers based on predefined request object, if needed
def _trigger_request(instance, request):
"""
Triggers request mock definition methods dynamically based on input
keyword arguments passed to `pook.Mock` constructor.
This is used to provide a more Pythonic interface vs chainable API
approach.
"""
if not isinstance(request, Request):
raise TypeError('request must be instance of pook.Request')
# Register request matchers
for key in request.keys:
if hasattr(instance, key):
getattr(instance, key)(getattr(request, key))