import functools
from inspect import isfunction, ismethod
from furl import furl
from .constants import TYPES
from .helpers import trigger_methods
from .matcher import MatcherEngine
from .matchers import init as matcher
from .request import Request
from .response import Response
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:
"""
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):
"""
Defines the body data to match.
``body`` argument can be a ``str``, ``bytes`` or a regular expression.
Arguments:
body (str|bytes|regex): body data to match.
Returns:
self: current Mock instance.
"""
if hasattr(body, "encode"):
body = body.encode("utf-8", "backslashreplace")
self._request.body = body
self.add_matcher(matcher("BodyMatcher", body))
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, "rb") as f:
return self.body(f.read())
[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.
"""
# 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
if self._times <= 0:
return False, [f"Mock matches request but is expired.\n{self!r}"]
# 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(f"{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