Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_getresponse(self):
entry = MocketEntry(('localhost', 8080), ['Show me.\r\n'])
self.assertEqual(entry.get_response(), encode_to_bytes('Show me.\r\n'))
def test_empty_getresponse(self):
entry = MocketEntry(('localhost', 8080), [])
self.assertEqual(entry.get_response(), encode_to_bytes(''))
def __init__(self, data):
_, self.body = decode_from_bytes(data).split('\r\n\r\n', 1)
self.rfile = BytesIO(encode_to_bytes(data))
self.raw_requestline = self.rfile.readline()
self.error_code = self.error_message = None
self.parse_request()
self.method = self.command
self.querystring = parse_qs(unquote_utf8(urlsplit(self.path).query), keep_blank_values=True)
def _hash_request(h, req):
return h(encode_to_bytes("".join(sorted(req.split("\r\n"))))).hexdigest()
raise
except KeyError:
# preventing next KeyError exceptions
responses.setdefault(self._host, dict())
responses[self._host].setdefault(port, dict())
responses[self._host][port].setdefault(req_signature, dict())
response_dict = responses[self._host][port][req_signature]
# try to get the response from the dictionary
try:
try:
encoded_response = hexdump.dehex(response_dict["response"])
except TypeError: # pragma: no cover
# Python 2
encoded_response = hexdump.restore(
encode_to_bytes(response_dict["response"])
)
# if not available, call the real sendall
except KeyError:
host, port = Mocket._address
host = true_gethostbyname(host)
if isinstance(self.true_socket, true_socket) and self._secure_socket:
try:
self = MocketSocket(sock=self)
except TypeError:
ssl_context = self.kwargs.get("ssl_context")
server_hostname = self.kwargs.get("server_hostname")
self.true_socket = true_ssl_context.wrap_socket(
self=ssl_context,
sock=self.true_socket,
server_hostname=server_hostname,
def tokens(iterable):
iterable = [encode_to_bytes(x) for x in iterable]
return [
'*{0}'.format(len(iterable)).encode('utf-8')
] + list(
chain(*zip(['${0}'.format(len(x)).encode('utf-8') for x in iterable], iterable))
)
def __getattr__(self, name):
if name == "last_request":
last_request = getattr(Mocket, "last_request")()
last_request.body = encode_to_bytes(last_request.body)
return last_request
elif name == "latest_requests":
return getattr(Mocket, "_requests")
else:
return getattr(Entry, name)
if not isinstance(responses, collections.Iterable) or isinstance(
responses, basestring
):
responses = [responses]
lresponses = []
for r in responses:
if not getattr(r, "data", False):
if isinstance(r, text_type):
r = encode_to_bytes(r)
r = self.response_cls(r)
lresponses.append(r)
else:
if not responses:
lresponses = [self.response_cls(encode_to_bytes(""))]
self.responses = lresponses