import functools
from furl import furl
from inspect import isfunction, ismethod
from .response import Response
from .constants import TYPES
from .request import Request
from .matcher import MatcherEngine
from .helpers import trigger_methods
from .exceptions import PookExpiredMock
from .matchers import init as matcher
def _append_funcs(target, items):
"""
Helper function to append functions into a given list.
Arguments:
target (list): receptor list to append functions.
items (iterable): iterable that yields elements to append.
"""
[target.append(item) for item in items if isfunction(item) or ismethod(item)]
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))
[docs]
class Mock(object):
"""
Mock is used to declare and compose the HTTP request/response mock
definition and matching expectations, which provides fluent API DSL.
Arguments:
url (str): URL to match.
E.g: ``server.com/api?foo=bar``.
method (str): HTTP method name to match.
E.g: ``GET``.
path (str): URL path to match.
E.g: ``/api/users``.
headers (dict): Header values to match.
E.g: ``{'server': 'nginx'}``.
header_present (str): Matches is a header is present.
headers_present (list|tuple): Matches if multiple headers are present.
type (str): Matches MIME ``Content-Type`` header.
E.g: ``json``, ``xml``, ``html``, ``text/plain``
content (str): Same as ``type`` argument.
params (dict): Matches the given URL params.
param_exists (str): Matches if a given URL param exists.
params_exists (list|tuple): Matches if a given URL params exists.
body (str|regex): Matches the payload body by regex or
strict comparison.
json (dict|list|str|regex): Matches the payload body against the given
JSON or regular expression.
jsonschema (dict|str): Matches the payload body against the given
JSONSchema.
xml (str|regex): matches the payload body against the given XML string
or regular expression.
file (str): Disk file path to load body from. Analog to ``body`` param.
times (int): Mock TTL or maximum number of times that the mock can be
matched.
persist (bool): Enable persistent mode. Mock won't be flushed even if
it matched one or multiple times.
delay (int): Optional network delay simulation (only applicable when
using ``aiohttp`` HTTP client).
callback (function): optional callback function called every time the
mock is matched.
reply (int): Mock response status. Defaults to ``200``.
response_status (int): Mock response status. Alias to ``reply`` param.
response_headers (dict): Response headers to use.
response_type (str): Response MIME type expression or alias.
Analog to ``type`` param. E.g: ``json``, ``xml``, ``text/plain``.
response_body (str): Response body to use.
response_json (dict|list|str): Response JSON to use. If Python is
passed, it will be serialized as JSON transparently.
response_xml (str): XML body string to use.
request (pook.Request): Optional. Request mock definition object.
response (pook.Response): Optional. Response mock definition
object.
Returns:
pook.Mock
"""
_KEY_ORDER = (
"add_matcher",
"body",
"callback",
"calls",
"content",
"delay",
"done",
"error",
"file",
"filter",
"header",
"header_present",
"headers",
"headers_present",
"isdone",
"ismatched",
"json",
"jsonschema",
"map",
"match",
"matched",
"matches",
"method",
"url",
"param",
"param_exists",
"params",
"path",
"persist",
"reply",
"response",
"status",
"times",
"total_matches",
"type",
"use",
"xml",
)
def __init__(self, request=None, response=None, **kw):
# 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, self._KEY_ORDER)
# Trigger matchers based on predefined request object, if needed
if request:
_trigger_request(self, request)
[docs]
def url(self, url):
"""
Defines the mock URL to match.
It can be a full URL with path and query params.
Protocol schema is optional, defaults to ``http://``.
Arguments:
url (str): mock URL to match. E.g: ``server.com/api``.
Returns:
self: current Mock instance.
"""
self._request.url = url
self.add_matcher(matcher("URLMatcher", url))
return self
[docs]
def method(self, method):
"""
Defines the HTTP method to match.
Use ``*`` to match any method.
Arguments:
method (str): method value to match. E.g: ``GET``.
Returns:
self: current Mock instance.
"""
self._request.method = method
self.add_matcher(matcher("MethodMatcher", method))
return self
[docs]
def path(self, path):
"""
Defines a URL path to match.
Only call this method if the URL has no path already defined.
Arguments:
path (str): URL path value to match. E.g: ``/api/users``.
Returns:
self: current Mock instance.
"""
url = furl(self._request.rawurl)
url.path = path
self._request.url = url.url
self.add_matcher(matcher("PathMatcher", path))
return self
[docs]
def type(self, value):
"""
Defines the request ``Content-Type`` header to match.
You can pass one of the following aliases instead of the full
MIME type representation:
- ``json`` = ``application/json``
- ``xml`` = ``application/xml``
- ``html`` = ``text/html``
- ``text`` = ``text/plain``
- ``urlencoded`` = ``application/x-www-form-urlencoded``
- ``form`` = ``application/x-www-form-urlencoded``
- ``form-data`` = ``application/x-www-form-urlencoded``
Arguments:
value (str): type alias or header value to match.
Returns:
self: current Mock instance.
"""
self.content(value)
return self
[docs]
def content(self, value):
"""
Defines the ``Content-Type`` outgoing header value to match.
You can pass one of the following type aliases instead of the full
MIME type representation:
- ``json`` = ``application/json``
- ``xml`` = ``application/xml``
- ``html`` = ``text/html``
- ``text`` = ``text/plain``
- ``urlencoded`` = ``application/x-www-form-urlencoded``
- ``form`` = ``application/x-www-form-urlencoded``
- ``form-data`` = ``application/x-www-form-urlencoded``
Arguments:
value (str): type alias or header value to match.
Returns:
self: current Mock instance.
"""
header = {"Content-Type": TYPES.get(value, value)}
self._request.headers = header
self.add_matcher(matcher("HeadersMatcher", header))
return self
[docs]
def param(self, name, value):
"""
Defines an URL param key and value to match.
Arguments:
name (str): param name value to match.
value (str): param name value to match.
Returns:
self: current Mock instance.
"""
self.params({name: value})
return self
[docs]
def param_exists(self, name, allow_empty=False):
"""
Checks if a given URL param name is present in the URL.
Arguments:
name (str): param name to check existence.
allow_empty (bool): whether to allow an empty value of the param
Returns:
self: current Mock instance.
"""
self.add_matcher(matcher("QueryParameterExistsMatcher", name, allow_empty))
return self
[docs]
def params(self, params):
"""
Defines a set of URL query params to match.
Arguments:
params (dict): set of params to match.
Returns:
self: current Mock instance.
"""
url = furl(self._request.rawurl)
url = url.add(params)
self._request.url = url.url
self.add_matcher(matcher("QueryMatcher", params))
return self
[docs]
def body(self, body, binary=False):
"""
Defines the body data to match.
``body`` argument can be a ``str``, ``binary`` or a regular expression.
Arguments:
body (str|binary|regex): body data to match.
binary (bool): prevent decoding body as text when True.
Returns:
self: current Mock instance.
"""
self._request.body = body
self.add_matcher(matcher("BodyMatcher", body, binary=False))
return self
[docs]
def json(self, json):
"""
Defines the JSON body to match.
``json`` argument can be an JSON string, a JSON serializable
Python structure, such as a ``dict`` or ``list`` or it can be
a regular expression used to match the body.
Arguments:
json (str|dict|list|regex): body JSON to match.
Returns:
self: current Mock instance.
"""
self._request.json = json
self.add_matcher(matcher("JSONMatcher", json))
return self
[docs]
def jsonschema(self, schema):
"""
Defines a JSONSchema representation to be used for body matching.
Arguments:
schema (str|dict): dict or JSONSchema string to use.
Returns:
self: current Mock instance.
"""
self.add_matcher(matcher("JSONSchemaMatcher", schema))
return self
[docs]
def xml(self, xml):
"""
Defines a XML body value to match.
Arguments:
xml (str|regex): body XML to match.
Returns:
self: current Mock instance.
"""
self._request.xml = xml
self.add_matcher(matcher("XMLMatcher", xml))
return self
[docs]
def file(self, path):
"""
Reads the body to match from a disk file.
Arguments:
path (str): relative or absolute path to file to read from.
Returns:
self: current Mock instance.
"""
with open(path, "r") as f:
self.body(str(f.read()))
return self
[docs]
def add_matcher(self, matcher):
"""
Adds one or multiple custom matchers instances.
Matchers must implement the following interface:
- ``.__init__(expectation)``
- ``.match(request)``
- ``.name = str``
Matchers can optionally inherit from ``pook.matchers.BaseMatcher``.
Arguments:
*matchers (pook.matchers.BaseMatcher): matchers to add.
Returns:
self: current Mock instance.
"""
self.matchers.add(matcher)
return self
[docs]
def use(self, *matchers):
"""
Adds one or multiple custom matchers instances.
Matchers must implement the following interface:
- ``.__init__(expectation)``
- ``.match(request)``
- ``.name = str``
Matchers can optionally inherit from ``pook.matchers.BaseMatcher``.
Arguments:
*matchers (pook.matchers.BaseMatcher): matchers to add.
Returns:
self: current Mock instance.
"""
[self.add_matcher(matcher) for matcher in matchers]
return self
[docs]
def times(self, times=1):
"""
Defines the TTL limit for the current mock.
The TTL number will determine the maximum number of times that the
current mock can be matched and therefore consumed.
Arguments:
times (int): TTL number. Defaults to ``1``.
Returns:
self: current Mock instance.
"""
self._times = times
return self
[docs]
def persist(self, status=None):
"""
Enables persistent mode for the current mock.
Returns:
self: current Mock instance.
"""
self._persist = status if isinstance(status, bool) else True
return self
[docs]
def filter(self, *filters):
"""
Registers one o multiple request filters used during the matching
phase.
Arguments:
*mappers (function): variadic mapper functions.
Returns:
self: current Mock instance.
"""
_append_funcs(self.filters, filters)
return self
[docs]
def map(self, *mappers):
"""
Registers one o multiple request mappers used during the mapping
phase.
Arguments:
*mappers (function): variadic mapper functions.
Returns:
self: current Mock instance.
"""
_append_funcs(self.mappers, mappers)
return self
[docs]
def callback(self, *callbacks):
"""
Registers one or multiple callback that will be called every time the
current mock matches an outgoing HTTP request.
Arguments:
*callbacks (function): callback functions to call.
Returns:
self: current Mock instance.
"""
_append_funcs(self.callbacks, callbacks)
return self
[docs]
def delay(self, delay=1000):
"""
Delay network response with certain milliseconds.
Only supported by asynchronous HTTP clients, such as ``aiohttp``.
Arguments:
delay (int): milliseconds to delay response.
Returns:
self: current Mock instance.
"""
self._delay = int(delay)
return self
[docs]
def error(self, error):
"""
Defines a simulated exception error that will be raised.
Arguments:
error (str|Exception): error to raise.
Returns:
self: current Mock instance.
"""
self._error = RuntimeError(error) if isinstance(error, str) else error
return self
[docs]
def reply(self, status=200, new_response=False, **kw):
"""
Defines the mock response.
Arguments:
status (int, optional): response status code. Defaults to ``200``.
**kw (dict): optional keyword arguments passed to ``pook.Response``
constructor.
Returns:
pook.Response: mock response definition instance.
"""
# Use or create a Response mock instance
res = Response(**kw) if new_response else self._response
# Define HTTP mandatory response status
res.status(status or res._status)
# Expose current mock instance in response for self-reference
res.mock = self
# Define mock response
self._response = res
# Return response
return res
[docs]
def status(self, code=200):
"""
Defines the response status code.
Equivalent to ``self.reply(code)``.
Arguments:
code (int): response status code. Defaults to ``200``.
Returns:
pook.Response: mock response definition instance.
"""
return self.reply(status=code)
[docs]
def response(self, status=200, **kw):
"""
Defines the mock response. Alias to ``.reply()``
Arguments:
status (int): response status code. Defaults to ``200``.
**kw (dict): optional keyword arguments passed to ``pook.Response``
constructor.
Returns:
pook.Response: mock response definition instance.
"""
return self.reply(status=status, **kw)
[docs]
def isdone(self):
"""
Returns ``True`` if the mock has been matched by outgoing HTTP traffic.
Returns:
bool: ``True`` if the mock was matched succesfully.
"""
return (self._persist and self._matches > 0) or self._times <= 0
[docs]
def ismatched(self):
"""
Returns ``True`` if the mock has been matched at least once time.
Returns:
bool
"""
return self._matches > 0
@property
def done(self):
"""
Attribute accessor that would be ``True`` if the current mock
is done, and therefore have been matched multiple times.
Returns:
bool
"""
return self.isdone()
@property
def matched(self):
"""
Accessor property that would be ``True`` if the current mock
have been matched at least once.
See ``Mock.total_matches`` for more information.
Returns:
bool
"""
return self._matches > 0
@property
def total_matches(self):
"""
Accessor property to retrieve the total number of times that the
current mock has been matched.
Returns:
int
"""
return self._matches
@property
def matches(self):
"""
Accessor to retrieve the mock match calls registry.
Returns:
list[MockCall]
"""
return self._calls
@property
def calls(self):
"""
Accessor to retrieve the amount of mock matched calls.
Returns:
int
"""
return len(self.matches)
[docs]
def match(self, request):
"""
Matches an outgoing HTTP request against the current mock matchers.
This method acts like a delegator to `pook.MatcherEngine`.
Arguments:
request (pook.Request): request instance to match.
Raises:
Exception: if the mock has an exception defined.
Returns:
tuple(bool, list[Exception]): ``True`` if the mock matches
the outgoing HTTP request, otherwise ``False``. Also returns
an optional list of error exceptions.
"""
# If mock already expired, fail it
if self._times <= 0:
raise PookExpiredMock("Mock expired")
# Trigger mock filters
for test in self.filters:
if not test(request, self):
return False, []
# Trigger mock mappers
for mapper in self.mappers:
request = mapper(request, self)
if not request:
raise ValueError("map function must return a request object")
# Match incoming request against registered mock matchers
matches, errors = self.matchers.match(request)
# If not matched, return False
if not matches:
return False, errors
# Register matched request for further inspecion and reference
self._calls.append(request)
# Increase mock call counter
self._matches += 1
if not self._persist:
self._times -= 1
# Raise simulated error
if self._error:
raise self._error
# Trigger callback when matched
for callback in self.callbacks:
callback(request, self)
return True, []
def __call__(self, fn):
"""
Overload Mock instance as callable object in order to be used
as decorator definition syntax.
Arguments:
fn (function): function to decorate.
Returns:
function or pook.Mock
"""
# Support chain sequences of mock definitions
if isinstance(fn, Response):
return fn.mock
if isinstance(fn, Mock):
return fn
# Force type assertion and raise an error if it is not a function
if not isfunction(fn) and not ismethod(fn):
raise TypeError("first argument must be a method or function")
# Remove mock to prevent decorator definition scope collision
self._engine.remove_mock(self)
@functools.wraps(fn)
def decorator(*args, **kw):
# Re-register mock on decorator call
self._engine.add_mock(self)
# Force engine activation, if available
# This prevents state issue while declaring mocks as decorators.
# This might be removed in the future.
engine_active = self._engine.active
if not engine_active:
self._engine.activate()
# Call decorated target function
try:
return fn(*args, **kw)
finally:
# Finally remove mock after function execution
# to prevent shared state
self._engine.remove_mock(self)
# If the engine was not previously active, disable it
if not engine_active:
self._engine.disable()
return decorator
def __repr__(self):
"""
Returns an human friendly readable instance data representation.
Returns:
str
"""
keys = ("matches", "times", "persist", "matchers", "response")
args = []
for key in keys:
if key == "matchers":
value = repr(self.matchers).replace("\n ", "\n ")
value = value[:-2] + " ])"
elif key == "response":
value = repr(self._response)
value = value[:-1] + " )"
else:
value = repr(getattr(self, "_" + key))
args.append("{}={}".format(key, value))
args = "(\n {}\n)".format(",\n ".join(args))
return type(self).__name__ + args
def __enter__(self):
"""
Implements context manager enter interface.
"""
# Make mock persistent if using default times
if self._times == 1:
self._persist = True
# Automatically enable the mock engine, if needed
if not self._engine.active:
self._engine.activate()
self._disable_engine = True
return self
def __exit__(self, etype, value, traceback):
"""
Implements context manager exit interface.
"""
# Force disable mock
self._times = 0
# Automatically disable the mock engine, if needed
if getattr(self, "_disable_engine", False):
self._disable_engine = False
self._engine.disable()
if etype is not None:
raise value