FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

gh-159690: Fix timed log rotation after a failed or duplicate rollover by lpyu001 · Pull Request #159692 · python/cpython · GitHub

Repository navigation

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (2) .rst  (1) All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
2 changes: 1 addition & 1 deletion Lib/logging/handlers.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
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

Back | FazBrowse Home | New Git URL