(self)
| 769 | self.assertEqual(d.unused_data + bi.read(), TRAIL) |
| 770 | |
| 771 | def test_decompressor_chunks_read_3(self): |
| 772 | TRAIL = b'89234893' |
| 773 | DAT = DAT_130K_C + TRAIL |
| 774 | d = ZstdDecompressor() |
| 775 | |
| 776 | bi = io.BytesIO(DAT) |
| 777 | lst = [] |
| 778 | while True: |
| 779 | if d.needs_input: |
| 780 | dat = bi.read(3) |
| 781 | if not dat: |
| 782 | break |
| 783 | else: |
| 784 | dat = b'' |
| 785 | |
| 786 | ret = d.decompress(dat, 1) |
| 787 | lst.append(ret) |
| 788 | if d.eof: |
| 789 | break |
| 790 | |
| 791 | ret = b''.join(lst) |
| 792 | |
| 793 | self.assertEqual(len(ret), _130_1K) |
| 794 | self.assertTrue(d.eof) |
| 795 | self.assertFalse(d.needs_input) |
| 796 | self.assertEqual(d.unused_data + bi.read(), TRAIL) |
| 797 | |
| 798 | |
| 799 | def test_decompress_empty(self): |
nothing calls this directly
no test coverage detected