A wrapper around BytesIO that restricts what can be read since data from the network can't be sought and cannot be read outside of its content length. This makes sure that views can't do anything under the test client that wouldn't work in real life.
| 66 | |
| 67 | |
| 68 | class FakePayload(IOBase): |
| 69 | """ |
| 70 | A wrapper around BytesIO that restricts what can be read since data from |
| 71 | the network can't be sought and cannot be read outside of its content |
| 72 | length. This makes sure that views can't do anything under the test client |
| 73 | that wouldn't work in real life. |
| 74 | """ |
| 75 | |
| 76 | def __init__(self, initial_bytes=None): |
| 77 | self.__content = BytesIO() |
| 78 | self.__len = 0 |
| 79 | self.read_started = False |
| 80 | if initial_bytes is not None: |
| 81 | self.write(initial_bytes) |
| 82 | |
| 83 | def __len__(self): |
| 84 | return self.__len |
| 85 | |
| 86 | def read(self, size=-1, /): |
| 87 | if not self.read_started: |
| 88 | self.__content.seek(0) |
| 89 | self.read_started = True |
| 90 | if size == -1 or size is None: |
| 91 | size = self.__len |
| 92 | assert ( |
| 93 | self.__len >= size |
| 94 | ), "Cannot read more than the available bytes from the HTTP incoming data." |
| 95 | content = self.__content.read(size) |
| 96 | self.__len -= len(content) |
| 97 | return content |
| 98 | |
| 99 | def readline(self, size=-1, /): |
| 100 | if not self.read_started: |
| 101 | self.__content.seek(0) |
| 102 | self.read_started = True |
| 103 | if size == -1 or size is None: |
| 104 | size = self.__len |
| 105 | assert ( |
| 106 | self.__len >= size |
| 107 | ), "Cannot read more than the available bytes from the HTTP incoming data." |
| 108 | content = self.__content.readline(size) |
| 109 | self.__len -= len(content) |
| 110 | return content |
| 111 | |
| 112 | def write(self, b, /): |
| 113 | if self.read_started: |
| 114 | raise ValueError("Unable to write a payload after it's been read") |
| 115 | content = force_bytes(b) |
| 116 | self.__content.write(content) |
| 117 | self.__len += len(content) |
| 118 | |
| 119 | |
| 120 | def closing_iterator_wrapper(iterable, close): |
no outgoing calls