Skip to content

Commit

Permalink
Replace stat.ST_xxx usage with os.stat().st_xxx (#116501)
Browse files Browse the repository at this point in the history
Modernize code to use the new API which avoids the usage of the stat
module just to read os.stat() members.

* Sort logging.handlers imports.
* Rework reopenIfNeeded() code to make it easier to follow.
* Replace "not self.stream" with "self.stream is None".
  • Loading branch information
vstinner authored Mar 8, 2024
1 parent cca3023 commit 6183158
Show file tree
Hide file tree
Showing 3 changed files with 40 additions and 27 deletions.
55 changes: 35 additions & 20 deletions Lib/logging/handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,17 @@
To use, simply 'import logging.handlers' and log away!
"""

import io, logging, socket, os, pickle, struct, time, re
from stat import ST_DEV, ST_INO, ST_MTIME
import copy
import io
import logging
import os
import pickle
import queue
import re
import socket
import struct
import threading
import copy
import time

#
# Some constants...
Expand Down Expand Up @@ -269,7 +275,7 @@ def __init__(self, filename, when='h', interval=1, backupCount=0,
# path object (see Issue #27493), but self.baseFilename will be a string
filename = self.baseFilename
if os.path.exists(filename):
t = os.stat(filename)[ST_MTIME]
t = int(os.stat(filename).st_mtime)
else:
t = int(time.time())
self.rolloverAt = self.computeRollover(t)
Expand Down Expand Up @@ -459,8 +465,7 @@ class WatchedFileHandler(logging.FileHandler):
This handler is not appropriate for use under Windows, because
under Windows open files cannot be moved or renamed - logging
opens the files with exclusive locks - and so there is no need
for such a handler. Furthermore, ST_INO is not supported under
Windows; stat always returns zero for this value.
for such a handler.
This handler is based on a suggestion and patch by Chad J.
Schroeder.
Expand All @@ -476,9 +481,11 @@ def __init__(self, filename, mode='a', encoding=None, delay=False,
self._statstream()

def _statstream(self):
if self.stream:
sres = os.fstat(self.stream.fileno())
self.dev, self.ino = sres[ST_DEV], sres[ST_INO]
if self.stream is None:
return
sres = os.fstat(self.stream.fileno())
self.dev = sres.st_dev
self.ino = sres.st_ino

def reopenIfNeeded(self):
"""
Expand All @@ -488,25 +495,33 @@ def reopenIfNeeded(self):
has, close the old stream and reopen the file to get the
current stream.
"""
if self.stream is None:
return

# Reduce the chance of race conditions by stat'ing by path only
# once and then fstat'ing our new fd if we opened a new log stream.
# See issue #14632: Thanks to John Mulligan for the problem report
# and patch.
try:
# stat the file by path, checking for existence
sres = os.stat(self.baseFilename)

# compare file system stat with that of our stream file handle
reopen = (sres.st_dev != self.dev or sres.st_ino != self.ino)
except FileNotFoundError:
sres = None
# compare file system stat with that of our stream file handle
if not sres or sres[ST_DEV] != self.dev or sres[ST_INO] != self.ino:
if self.stream is not None:
# we have an open file handle, clean it up
self.stream.flush()
self.stream.close()
self.stream = None # See Issue #21742: _open () might fail.
# open a new file handle and get new stat info from that fd
self.stream = self._open()
self._statstream()
reopen = True

if not reopen:
return

# we have an open file handle, clean it up
self.stream.flush()
self.stream.close()
self.stream = None # See Issue #21742: _open () might fail.

# open a new file handle and get new stat info from that fd
self.stream = self._open()
self._statstream()

def emit(self, record):
"""
Expand Down
9 changes: 4 additions & 5 deletions Lib/test/test_largefile.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
"""

import os
import stat
import sys
import unittest
import socket
Expand All @@ -29,7 +28,7 @@ def setUp(self):
mode = 'w+b'

with self.open(TESTFN, mode) as f:
current_size = os.fstat(f.fileno())[stat.ST_SIZE]
current_size = os.fstat(f.fileno()).st_size
if current_size == size+1:
return

Expand All @@ -40,13 +39,13 @@ def setUp(self):
f.seek(size)
f.write(b'a')
f.flush()
self.assertEqual(os.fstat(f.fileno())[stat.ST_SIZE], size+1)
self.assertEqual(os.fstat(f.fileno()).st_size, size+1)

@classmethod
def tearDownClass(cls):
with cls.open(TESTFN, 'wb'):
pass
if not os.stat(TESTFN)[stat.ST_SIZE] == 0:
if not os.stat(TESTFN).st_size == 0:
raise cls.failureException('File was not truncated by opening '
'with mode "wb"')
unlink(TESTFN2)
Expand All @@ -67,7 +66,7 @@ def test_large_read(self, _size):
self.assertEqual(f.tell(), size + 1)

def test_osstat(self):
self.assertEqual(os.stat(TESTFN)[stat.ST_SIZE], size+1)
self.assertEqual(os.stat(TESTFN).st_size, size+1)

def test_seek_read(self):
with self.open(TESTFN, 'rb') as f:
Expand Down
3 changes: 1 addition & 2 deletions Lib/test/test_mailbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -702,8 +702,7 @@ def _check_basics(self, factory=None):
self.assertEqual(self._box._factory, factory)
for subdir in '', 'tmp', 'new', 'cur':
path = os.path.join(self._path, subdir)
mode = os.stat(path)[stat.ST_MODE]
self.assertTrue(stat.S_ISDIR(mode), "Not a directory: '%s'" % path)
self.assertTrue(os.path.isdir(path), f"Not a directory: {path!r}")

def test_list_folders(self):
# List folders
Expand Down

0 comments on commit 6183158

Please sign in to comment.