import json as _json
from .regex import isregex
from .headers import HTTPHeaderDict
from .helpers import trigger_methods
from .matchers.url import protoregex
from urllib.parse import urlparse, parse_qs, urlunparse
[docs]
class Request(object):
"""
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 (str|regex): request body payload to match.
json (str|dict|list): JSON payload body structure to match.
xml (str): XML payload data structure to match.
Attributes:
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 (str|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")
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):
return self._method
@method.setter
def method(self, method):
self._method = method
@property
def headers(self):
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):
return self._url
@property
def rawurl(self):
return self._url if isregex(self._url) else urlunparse(self._url)
@url.setter
def url(self, url):
if isregex(url):
self._url = url
else:
if not protoregex.match(url):
url = "http://{}".format(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 query(self):
return self._query
@query.setter
def query(self, params):
self._query = parse_qs(params)
@property
def body(self):
return self._body
@body.setter
def body(self, body):
if hasattr(body, "decode"):
try:
body = body.decode("utf-8", "strict")
except Exception:
pass
self._body = body
@property
def json(self):
return _json.loads(self._body)
@json.setter
def json(self, data):
if isinstance(data, str):
self._body = data
else:
self._body = _json.dumps(data)
@property
def xml(self):
return self._body
@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("Method: {}".format(self._method))
entries.append(
"URL: {}".format(self._url if isregex(self._url) else self.rawurl)
)
if self._query:
entries.append("Query: {}".format(self._query))
if self._headers:
entries.append("Headers: {}".format(self._headers))
if self._body:
entries.append("Body: {}".format(self._body))
separator = "=" * 50
return (separator + "\n{}\n" + separator).format("\n".join(entries))