(self, f, compression)
| 33 | self.data = '\n'.join(line_gen).encode('ascii') |
| 34 | |
| 35 | def zipTest(self, f, compression): |
| 36 | # Create the ZIP archive. |
| 37 | with zipfile.ZipFile(f, "w", compression) as zipfp: |
| 38 | |
| 39 | # It will contain enough copies of self.data to reach about 6 GiB of |
| 40 | # raw data to store. |
| 41 | filecount = 6*1024**3 // len(self.data) |
| 42 | |
| 43 | next_time = time.monotonic() + _PRINT_WORKING_MSG_INTERVAL |
| 44 | for num in range(filecount): |
| 45 | zipfp.writestr("testfn%d" % num, self.data) |
| 46 | # Print still working message since this test can be really slow |
| 47 | if next_time <= time.monotonic(): |
| 48 | next_time = time.monotonic() + _PRINT_WORKING_MSG_INTERVAL |
| 49 | print(( |
| 50 | ' zipTest still writing %d of %d, be patient...' % |
| 51 | (num, filecount)), file=sys.__stdout__) |
| 52 | sys.__stdout__.flush() |
| 53 | |
| 54 | # Read the ZIP archive |
| 55 | with zipfile.ZipFile(f, "r", compression) as zipfp: |
| 56 | for num in range(filecount): |
| 57 | self.assertEqual(zipfp.read("testfn%d" % num), self.data) |
| 58 | # Print still working message since this test can be really slow |
| 59 | if next_time <= time.monotonic(): |
| 60 | next_time = time.monotonic() + _PRINT_WORKING_MSG_INTERVAL |
| 61 | print(( |
| 62 | ' zipTest still reading %d of %d, be patient...' % |
| 63 | (num, filecount)), file=sys.__stdout__) |
| 64 | sys.__stdout__.flush() |
| 65 | |
| 66 | # Check that testzip thinks the archive is valid |
| 67 | self.assertIsNone(zipfp.testzip()) |
| 68 | |
| 69 | def testStored(self): |
| 70 | # Try the temp file first. If we do TESTFN2 first, then it hogs |
no test coverage detected