(self)
| 6506 | |
| 6507 | @support.requires_zlib() |
| 6508 | def test_rotator(self): |
| 6509 | def namer(name): |
| 6510 | return name + ".gz" |
| 6511 | |
| 6512 | def rotator(source, dest): |
| 6513 | with open(source, "rb") as sf: |
| 6514 | data = sf.read() |
| 6515 | compressed = zlib.compress(data, 9) |
| 6516 | with open(dest, "wb") as df: |
| 6517 | df.write(compressed) |
| 6518 | os.remove(source) |
| 6519 | |
| 6520 | rh = logging.handlers.RotatingFileHandler( |
| 6521 | self.fn, encoding="utf-8", backupCount=2, maxBytes=1) |
| 6522 | rh.rotator = rotator |
| 6523 | rh.namer = namer |
| 6524 | m1 = self.next_rec() |
| 6525 | rh.emit(m1) |
| 6526 | self.assertLogFile(self.fn) |
| 6527 | m2 = self.next_rec() |
| 6528 | rh.emit(m2) |
| 6529 | fn = namer(self.fn + ".1") |
| 6530 | self.assertLogFile(fn) |
| 6531 | newline = os.linesep |
| 6532 | with open(fn, "rb") as f: |
| 6533 | compressed = f.read() |
| 6534 | data = zlib.decompress(compressed) |
| 6535 | self.assertEqual(data.decode("ascii"), m1.msg + newline) |
| 6536 | rh.emit(self.next_rec()) |
| 6537 | fn = namer(self.fn + ".2") |
| 6538 | self.assertLogFile(fn) |
| 6539 | with open(fn, "rb") as f: |
| 6540 | compressed = f.read() |
| 6541 | data = zlib.decompress(compressed) |
| 6542 | self.assertEqual(data.decode("ascii"), m1.msg + newline) |
| 6543 | rh.emit(self.next_rec()) |
| 6544 | fn = namer(self.fn + ".2") |
| 6545 | with open(fn, "rb") as f: |
| 6546 | compressed = f.read() |
| 6547 | data = zlib.decompress(compressed) |
| 6548 | self.assertEqual(data.decode("ascii"), m2.msg + newline) |
| 6549 | self.assertFalse(os.path.exists(namer(self.fn + ".3"))) |
| 6550 | rh.close() |
| 6551 | |
| 6552 | class TimedRotatingFileHandlerTest(BaseFileTest): |
| 6553 | @unittest.skipIf(support.is_wasi, "WASI does not have /dev/null.") |
nothing calls this directly
no test coverage detected