(
self, get_func: typing.Callable[[BytesQueueBuffer], bytes]
)
| 129 | "12.5 MB", current_thread_only=True |
| 130 | ) # assert that we're not doubling memory usagelimit_mem |
| 131 | def test_memory_usage( |
| 132 | self, get_func: typing.Callable[[BytesQueueBuffer], bytes] |
| 133 | ) -> None: |
| 134 | # Allocate 10 1MiB chunks |
| 135 | buffer = BytesQueueBuffer() |
| 136 | for i in range(10): |
| 137 | # This allocates 2MiB, putting the max at around 12MiB. Not sure why. |
| 138 | buffer.put(bytes(2**20)) |
| 139 | |
| 140 | result = get_func(buffer) |
| 141 | assert type(result) is bytes |
| 142 | assert len(result) == 10 * 2**20 |
| 143 | |
| 144 | @pytest.mark.parametrize( |
| 145 | "get_func", |
nothing calls this directly
no test coverage detected