()
| 175 | |
| 176 | |
| 177 | def test_partial_cache_chunk_buffer(): |
| 178 | big = "0123456789abcdefghijklmnopqrstuvwxyz" * 25000 |
| 179 | data = [Item(path="full", target=big), Item(path="partial", target=big)] |
| 180 | cache = MockCache() |
| 181 | key = PlaintextKey(None) |
| 182 | chunks = CacheChunkBuffer(cache, key, None) |
| 183 | for d in data: |
| 184 | chunks.add(d) |
| 185 | chunks.flush(flush=False) |
| 186 | # the code is expected to leave the last partial chunk in the buffer |
| 187 | assert len(chunks.chunks) == 3 |
| 188 | assert chunks.buffer.tell() > 0 |
| 189 | # now really flush |
| 190 | chunks.flush(flush=True) |
| 191 | assert len(chunks.chunks) == 4 |
| 192 | assert chunks.buffer.tell() == 0 |
| 193 | unpacker = msgpack.Unpacker() |
| 194 | for id in chunks.chunks: |
| 195 | unpacker.feed(cache.objects[id]) |
| 196 | assert data == [Item(internal_dict=d) for d in unpacker] |
| 197 | |
| 198 | |
| 199 | def make_chunks(items): |
nothing calls this directly
no test coverage detected