Source code for pook.mock

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 header(self, name, value): """ 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. """ headers = {name: value} self._request.headers = headers self.add_matcher(matcher("HeadersMatcher", headers)) return self
[docs] def headers(self, headers=None, **kw): """ Defines a dictionary of arguments. Header keys are case insensitive. Arguments: headers (dict): headers to match. **headers (dict): headers to match as variadic keyword arguments. Returns: self: current Mock instance. """ headers = kw if kw else headers self._request.headers = headers self.add_matcher(matcher("HeadersMatcher", headers)) return self
[docs] def header_present(self, *names): """ Defines a new header matcher expectation that must be present in the outgoing request in order to be satisfied, no matter what value it hosts. Header keys are case insensitive. Arguments: *names (str): header or headers names to match. Returns: self: current Mock instance. Example:: (pook.get('server.com/api') .header_present('content-type')) """ return self.headers_present(names)
[docs] def headers_present(self, headers): """ Defines a list of headers that must be present in the outgoing request in order to satisfy the matcher, no matter what value the headers hosts. Header keys are case insensitive. Arguments: headers (list|tuple): header keys to match. Returns: self: current Mock instance. Example:: (pook.get('server.com/api') .headers_present(['content-type', 'Authorization'])) """ if not headers: raise ValueError("`headers` must not be empty") for header in headers: self.add_matcher(matcher("HeaderExistsMatcher", header)) 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