(self)
| 366 | self.assertEqual(dat3, THIS_FILE_BYTES) |
| 367 | |
| 368 | def test_compress_flushframe(self): |
| 369 | # test compress & decompress |
| 370 | point = len(THIS_FILE_BYTES) // 2 |
| 371 | |
| 372 | c = ZstdCompressor() |
| 373 | |
| 374 | dat1 = c.compress(THIS_FILE_BYTES[:point]) |
| 375 | self.assertEqual(c.last_mode, c.CONTINUE) |
| 376 | |
| 377 | dat1 += c.compress(THIS_FILE_BYTES[point:], c.FLUSH_FRAME) |
| 378 | self.assertEqual(c.last_mode, c.FLUSH_FRAME) |
| 379 | |
| 380 | nt = get_frame_info(dat1) |
| 381 | self.assertEqual(nt.decompressed_size, None) # no content size |
| 382 | |
| 383 | dat2 = decompress(dat1) |
| 384 | |
| 385 | self.assertEqual(dat2, THIS_FILE_BYTES) |
| 386 | |
| 387 | # single .FLUSH_FRAME mode has content size |
| 388 | c = ZstdCompressor() |
| 389 | dat = c.compress(THIS_FILE_BYTES, mode=c.FLUSH_FRAME) |
| 390 | self.assertEqual(c.last_mode, c.FLUSH_FRAME) |
| 391 | |
| 392 | nt = get_frame_info(dat) |
| 393 | self.assertEqual(nt.decompressed_size, len(THIS_FILE_BYTES)) |
| 394 | |
| 395 | def test_compress_empty(self): |
| 396 | # output empty content frame |
nothing calls this directly
no test coverage detected