1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103
|
"""
Hooks for pytest-httpserver
"""
import os
import time
from typing import Callable
from werkzeug import Request
from werkzeug import Response
class Chain:
"""
Combine multiple hooks into one callable object
Hooks specified will be called one by one.
Each hook will receive the response object made by the previous hook,
similar to reduce.
"""
def __init__(self, *args: Callable[[Request, Response], Response]):
"""
:param *args: callable objects specified in the same order they should
be called.
"""
self._hooks = args
def __call__(self, request: Request, response: Response) -> Response:
"""
Calls the callable object one by one. The second and further callable
objects receive the response returned by the previous one, while the
first one receives the original response object.
"""
for hook in self._hooks:
response = hook(request, response)
return response
class Delay:
"""
Delays returning the response
"""
def __init__(self, seconds: float):
"""
:param seconds: seconds to sleep before returning the response
"""
self._seconds = seconds
def _sleep(self):
"""
Sleeps for the seconds specified in the constructor
"""
time.sleep(self._seconds)
def __call__(self, _request: Request, response: Response) -> Response:
"""
Delays returning the response object for the time specified in the
constructor. Returns the original response unmodified.
"""
self._sleep()
return response
class Garbage:
def __init__(self, prefix_size: int = 0, suffix_size: int = 0):
"""
Adds random bytes to the beginning or to the end of the response data.
:param prefix_size: amount of random bytes to be added to the beginning
of the response data
:param suffix_size: amount of random bytes to be added to the end
of the response data
"""
assert prefix_size >= 0, "prefix_size should be positive integer"
assert suffix_size >= 0, "suffix_size should be positive integer"
self._prefix_size = prefix_size
self._suffix_size = suffix_size
def _get_garbage_bytes(self, size: int) -> bytes:
"""
Returns the specified amount of random bytes.
:param size: amount of bytes to return
"""
return os.urandom(size)
def __call__(self, _request: Request, response: Response) -> Response:
"""
Adds random bytes to the beginning or to the end of the response data.
New random bytes will be generated for every call.
Returns the modified response object.
"""
prefix = self._get_garbage_bytes(self._prefix_size)
suffix = self._get_garbage_bytes(self._suffix_size)
response.set_data(prefix + response.get_data() + suffix)
return response
|