Source code for pook.request

import json as _json
from urllib.parse import parse_qs, urlparse, urlunparse

from .headers import HTTPHeaderDict
from .helpers import trigger_methods
from .matchers.url import protoregex
from .regex import isregex


[docs] class Request: """ Request object representing the request mock expectation DSL. Arguments: method (str): HTTP method to match. Defaults to ``GET``. url (str): URL request to intercept and match. headers (dict): HTTP headers to match. query (dict): URL query params to match. Complementely to URL defined query params. body (bytes|regex): request body payload to match. json (str|dict|list): JSON payload body structure to match. xml (str): XML payload data structure to match. """ # Store keys keys = ("method", "headers", "body", "url", "query", "xml", "json") """ :meta private: """ def __init__(self, method="GET", **kw): self._url = None self._body = None self._query = None self._method = method self._extra = kw.get("extra") self._headers = HTTPHeaderDict() trigger_methods(self, kw, self.keys) @property def method(self): """HTTP method to match. Defaults to ``GET``.""" return self._method @method.setter def method(self, method): self._method = method @property def headers(self): """HTTP headers to match.""" return self._headers @headers.setter def headers(self, headers): if not hasattr(headers, "__setitem__"): raise TypeError("headers must be a dictionary") self._headers.extend(headers) @property def extra(self): return self._extra @extra.setter def extra(self, extra): if not isinstance(extra, dict): raise TypeError("extra must be a dictionary") self._extra = extra @property def url(self): """URL request to intercept and match.""" return self._url @url.setter def url(self, url): if isregex(url): self._url = url else: if not protoregex.match(url): url = f"http://{url}" self._url = urlparse(url) # keep_blank_values necessary for `param_exists` when a parameter has no value but is present self._query = ( parse_qs(self._url.query, keep_blank_values=True) if self._url.query else self._query ) @property def rawurl(self): return self._url if isregex(self._url) else urlunparse(self._url) @property def query(self): """URL query params to match. Complementary to URL defined query params.""" return self._query @query.setter def query(self, params): self._query = parse_qs(params) @property def body(self): """request body payload to match.""" return self._body @body.setter def body(self, body): if hasattr(body, "encode"): body = body.encode("utf-8", "backslashreplace") self._body = body @property def json(self): """JSON payload body structure to match.""" return _json.loads(self.body.decode("utf-8")) @json.setter def json(self, data): if isinstance(data, str): self.body = data else: self.body = _json.dumps(data) @property def xml(self): """XML payload data structure to match.""" return self.body.decode("utf-8") @xml.setter def xml(self, data): self.body = data
[docs] def copy(self): """ Copies the current Request object instance for side-effects purposes. Returns: pook.Request: copy of the current Request instance. """ req = type(self)() req.__dict__ = self.__dict__.copy() req._headers = self.headers.copy() return req
def __repr__(self): """ Returns an human friendly readable instance data representation. Returns: str """ entries = [] entries.append(f"Method: {self._method}") entries.append(f"URL: {self._url if isregex(self._url) else self.rawurl}") if self._query: entries.append(f"Query: {self._query}") if self._headers: entries.append(f"Headers: {self._headers}") if self._body: entries.append(f"Body: {self._body}") separator = "=" * 50 return (separator + "\n{}\n" + separator).format("\n".join(entries))