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