if self.stream is None: # delay was set...
self.stream = self._open()
if self.maxBytes > 0: # are we rolling over?
- pos = self.stream.tell()
+ try:
+ pos = self.stream.tell()
+ except io.UnsupportedOperation:
+ # gh-143237: Never rollover a named pipe.
+ return False
if not pos:
# gh-116263: Never rollover an empty file
return False
import codecs
import configparser
+import contextlib
import copy
import datetime
import pathlib
self.assertFalse(rh.shouldRollover(self.next_rec()))
rh.close()
+ @unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()')
+ def test_should_not_rollover_named_pipe(self):
+ # gh-143237 - test with non-seekable special file (named pipe)
+ filename = os_helper.TESTFN
+ self.addCleanup(os_helper.unlink, filename)
+ try:
+ os.mkfifo(filename)
+ except PermissionError as e:
+ self.skipTest('os.mkfifo(): %s' % e)
+
+ data = 'not read'
+ def other_side():
+ nonlocal data
+ with open(filename, 'rb') as f:
+ data = f.read()
+
+ thread = threading.Thread(target=other_side)
+ with threading_helper.start_threads([thread]):
+ rh = logging.handlers.RotatingFileHandler(
+ filename, encoding="utf-8", maxBytes=1)
+ with contextlib.closing(rh):
+ m = self.next_rec()
+ self.assertFalse(rh.shouldRollover(m))
+ rh.emit(m)
+ self.assertEqual(data.decode(), m.msg + os.linesep)
+
def test_should_rollover(self):
with open(self.fn, 'wb') as f:
f.write(b'\n')