Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Lib/logging/handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -469,6 +469,7 @@ def doRollover(self):
timeTuple = time.localtime(t + addend)
dfn = self.rotation_filename(self.baseFilename + "." +
time.strftime(self.suffix, timeTuple))
self.rolloverAt = self.computeRollover(currentTime)
if os.path.exists(dfn):
# Already rolled over.
return
Expand All @@ -482,7 +483,6 @@ def doRollover(self):
os.remove(s)
if not self.delay:
self.stream = self._open()
self.rolloverAt = self.computeRollover(currentTime)

class WatchedFileHandler(logging.FileHandler):
"""
Expand Down
69 changes: 69 additions & 0 deletions Lib/test/test_logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -6902,6 +6902,75 @@ def test_rollover_at_midnight(self, weekly=False):
def test_rollover_at_weekday(self):
self.test_rollover_at_midnight(weekly=True)

def _make_fake_time_handler(self, now, when='S', **kwargs):
dirname = tempfile.mkdtemp()
self.addCleanup(os_helper.rmtree, dirname)
fn = os.path.join(dirname, 'test.log')
with patch('time.time', lambda: now[0]):
fh = logging.handlers.TimedRotatingFileHandler(
fn, when=when, encoding='utf-8', delay=True, **kwargs)
self.addCleanup(fh.close)
return fh

def test_rollover_after_failed_rollover(self):
for target in 'os.remove', '_open':
with self.subTest(target=target):
now = [1_000_000_000]
fh = self._make_fake_time_handler(now, backupCount=1)
fn = fh.baseFilename
emit = lambda msg: fh.emit(logging.makeLogRecord({'msg': msg}))
with patch('time.time', lambda: now[0]):
for msg in 'a', 'b':
now[0] += 2
emit(msg)
now[0] += 2
if target == '_open':
failure = patch.object(fh, '_open', side_effect=OSError)
fh.delay = False
else:
failure = patch('os.remove', side_effect=OSError)
with failure as failed, support.captured_stderr():
emit('c')
failed.assert_called_once()
self.assertGreater(fh.rolloverAt, now[0])
fh.delay = True
for msg in 'd', 'e':
now[0] += 2
emit(msg)
fh.close()
with open(fn, encoding='utf-8') as f:
self.assertEqual(f.read(), 'e\n')
backups = [f for f in os.listdir(os.path.dirname(fn))
if f != 'test.log']
self.assertEqual(len(backups), 1)

def test_rollover_existing_backup(self):
now = [1_000_000_000]
fh = self._make_fake_time_handler(now)
fn = fh.baseFilename
with patch('time.time', lambda: now[0]):
fh.emit(logging.makeLogRecord({'msg': 'a'}))
t = fh.rolloverAt - fh.interval
dfn = fh.rotation_filename(
fn + '.' + time.strftime(fh.suffix, time.localtime(t)))
with open(dfn, 'w', encoding='utf-8') as f:
f.write('existing\n')
now[0] += 2
fh.emit(logging.makeLogRecord({'msg': 'b'}))
self.assertGreater(fh.rolloverAt, now[0])
t = fh.rolloverAt - fh.interval
next_dfn = fh.rotation_filename(
fn + '.' + time.strftime(fh.suffix, time.localtime(t)))
now[0] += 2
fh.emit(logging.makeLogRecord({'msg': 'c'}))
fh.close()
with open(dfn, encoding='utf-8') as f:
self.assertEqual(f.read(), 'existing\n')
with open(next_dfn, encoding='utf-8') as f:
self.assertEqual(f.read(), 'a\nb\n')
with open(fn, encoding='utf-8') as f:
self.assertEqual(f.read(), 'c\n')

def test_invalid(self):
assertRaises = self.assertRaises
assertRaises(ValueError, logging.handlers.TimedRotatingFileHandler,
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
Fix :class:`logging.handlers.TimedRotatingFileHandler` permanently stopping
rollover after a rollover failed or its backup file already existed, for
example at a DST change.
Loading