881 lines
32 KiB
Python
881 lines
32 KiB
Python
import sys
|
|
import os
|
|
import unittest
|
|
import itertools
|
|
import select
|
|
import signal
|
|
import stat
|
|
import subprocess
|
|
import time
|
|
from array import array
|
|
from weakref import proxy
|
|
try:
|
|
import threading
|
|
except ImportError:
|
|
threading = None
|
|
|
|
from test import test_support
|
|
from test.test_support import TESTFN, run_unittest, requires
|
|
from UserList import UserList
|
|
|
|
class AutoFileTests(unittest.TestCase):
|
|
# file tests for which a test file is automatically set up
|
|
|
|
def setUp(self):
|
|
self.f = open(TESTFN, 'wb')
|
|
|
|
def tearDown(self):
|
|
if self.f:
|
|
self.f.close()
|
|
os.remove(TESTFN)
|
|
|
|
def testWeakRefs(self):
|
|
# verify weak references
|
|
p = proxy(self.f)
|
|
p.write('teststring')
|
|
self.assertEqual(self.f.tell(), p.tell())
|
|
self.f.close()
|
|
self.f = None
|
|
self.assertRaises(ReferenceError, getattr, p, 'tell')
|
|
|
|
def testAttributes(self):
|
|
# verify expected attributes exist
|
|
f = self.f
|
|
with test_support.check_py3k_warnings():
|
|
softspace = f.softspace
|
|
f.name # merely shouldn't blow up
|
|
f.mode # ditto
|
|
f.closed # ditto
|
|
|
|
with test_support.check_py3k_warnings():
|
|
# verify softspace is writable
|
|
f.softspace = softspace # merely shouldn't blow up
|
|
|
|
# verify the others aren't
|
|
for attr in 'name', 'mode', 'closed':
|
|
self.assertRaises((AttributeError, TypeError), setattr, f, attr, 'oops')
|
|
|
|
def testReadinto(self):
|
|
# verify readinto
|
|
self.f.write('12')
|
|
self.f.close()
|
|
a = array('c', 'x'*10)
|
|
self.f = open(TESTFN, 'rb')
|
|
n = self.f.readinto(a)
|
|
self.assertEqual('12', a.tostring()[:n])
|
|
|
|
def testWritelinesUserList(self):
|
|
# verify writelines with instance sequence
|
|
l = UserList(['1', '2'])
|
|
self.f.writelines(l)
|
|
self.f.close()
|
|
self.f = open(TESTFN, 'rb')
|
|
buf = self.f.read()
|
|
self.assertEqual(buf, '12')
|
|
|
|
def testWritelinesIntegers(self):
|
|
# verify writelines with integers
|
|
self.assertRaises(TypeError, self.f.writelines, [1, 2, 3])
|
|
|
|
def testWritelinesIntegersUserList(self):
|
|
# verify writelines with integers in UserList
|
|
l = UserList([1,2,3])
|
|
self.assertRaises(TypeError, self.f.writelines, l)
|
|
|
|
def testWritelinesNonString(self):
|
|
# verify writelines with non-string object
|
|
class NonString:
|
|
pass
|
|
|
|
self.assertRaises(TypeError, self.f.writelines,
|
|
[NonString(), NonString()])
|
|
|
|
def testWritelinesBuffer(self):
|
|
self.f.writelines([array('c', 'abc')])
|
|
self.f.close()
|
|
self.f = open(TESTFN, 'rb')
|
|
buf = self.f.read()
|
|
self.assertEqual(buf, 'abc')
|
|
|
|
def testRepr(self):
|
|
# verify repr works
|
|
self.assertTrue(repr(self.f).startswith("<open file '" + TESTFN))
|
|
# see issue #14161
|
|
# Windows doesn't like \r\n\t" in the file name, but ' is ok
|
|
fname = 'xx\rxx\nxx\'xx"xx' if sys.platform != "win32" else "xx'xx"
|
|
with open(fname, 'w') as f:
|
|
self.addCleanup(os.remove, fname)
|
|
self.assertTrue(repr(f).startswith(
|
|
"<open file %r, mode 'w' at" % fname))
|
|
|
|
def testErrors(self):
|
|
self.f.close()
|
|
self.f = open(TESTFN, 'rb')
|
|
f = self.f
|
|
self.assertEqual(f.name, TESTFN)
|
|
self.assertTrue(not f.isatty())
|
|
self.assertTrue(not f.closed)
|
|
|
|
self.assertRaises(TypeError, f.readinto, "")
|
|
f.close()
|
|
self.assertTrue(f.closed)
|
|
|
|
def testMethods(self):
|
|
methods = ['fileno', 'flush', 'isatty', 'next', 'read', 'readinto',
|
|
'readline', 'readlines', 'seek', 'tell', 'truncate',
|
|
'write', '__iter__']
|
|
deprecated_methods = ['xreadlines']
|
|
if sys.platform.startswith('atheos'):
|
|
methods.remove('truncate')
|
|
|
|
# __exit__ should close the file
|
|
self.f.__exit__(None, None, None)
|
|
self.assertTrue(self.f.closed)
|
|
|
|
for methodname in methods:
|
|
method = getattr(self.f, methodname)
|
|
# should raise on closed file
|
|
self.assertRaises(ValueError, method)
|
|
with test_support.check_py3k_warnings():
|
|
for methodname in deprecated_methods:
|
|
method = getattr(self.f, methodname)
|
|
self.assertRaises(ValueError, method)
|
|
self.assertRaises(ValueError, self.f.writelines, [])
|
|
|
|
# file is closed, __exit__ shouldn't do anything
|
|
self.assertEqual(self.f.__exit__(None, None, None), None)
|
|
# it must also return None if an exception was given
|
|
try:
|
|
1 // 0
|
|
except:
|
|
self.assertEqual(self.f.__exit__(*sys.exc_info()), None)
|
|
|
|
def testReadWhenWriting(self):
|
|
self.assertRaises(IOError, self.f.read)
|
|
|
|
def testNastyWritelinesGenerator(self):
|
|
def nasty():
|
|
for i in range(5):
|
|
if i == 3:
|
|
self.f.close()
|
|
yield str(i)
|
|
self.assertRaises(ValueError, self.f.writelines, nasty())
|
|
|
|
def testIssue5677(self):
|
|
# Remark: Do not perform more than one test per open file,
|
|
# since that does NOT catch the readline error on Windows.
|
|
data = 'xxx'
|
|
for mode in ['w', 'wb', 'a', 'ab']:
|
|
for attr in ['read', 'readline', 'readlines']:
|
|
self.f = open(TESTFN, mode)
|
|
self.f.write(data)
|
|
self.assertRaises(IOError, getattr(self.f, attr))
|
|
self.f.close()
|
|
|
|
self.f = open(TESTFN, mode)
|
|
self.f.write(data)
|
|
self.assertRaises(IOError, lambda: [line for line in self.f])
|
|
self.f.close()
|
|
|
|
self.f = open(TESTFN, mode)
|
|
self.f.write(data)
|
|
self.assertRaises(IOError, self.f.readinto, bytearray(len(data)))
|
|
self.f.close()
|
|
|
|
for mode in ['r', 'rb', 'U', 'Ub', 'Ur', 'rU', 'rbU', 'rUb']:
|
|
self.f = open(TESTFN, mode)
|
|
self.assertRaises(IOError, self.f.write, data)
|
|
self.f.close()
|
|
|
|
self.f = open(TESTFN, mode)
|
|
self.assertRaises(IOError, self.f.writelines, [data, data])
|
|
self.f.close()
|
|
|
|
self.f = open(TESTFN, mode)
|
|
self.assertRaises(IOError, self.f.truncate)
|
|
self.f.close()
|
|
|
|
class OtherFileTests(unittest.TestCase):
|
|
|
|
def testOpenDir(self):
|
|
this_dir = os.path.dirname(__file__) or os.curdir
|
|
for mode in (None, "w"):
|
|
try:
|
|
if mode:
|
|
f = open(this_dir, mode)
|
|
else:
|
|
f = open(this_dir)
|
|
except IOError as e:
|
|
self.assertEqual(e.filename, this_dir)
|
|
else:
|
|
self.fail("opening a directory didn't raise an IOError")
|
|
|
|
def testModeStrings(self):
|
|
# check invalid mode strings
|
|
for mode in ("", "aU", "wU+"):
|
|
try:
|
|
f = open(TESTFN, mode)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
f.close()
|
|
self.fail('%r is an invalid file mode' % mode)
|
|
|
|
# Some invalid modes fail on Windows, but pass on Unix
|
|
# Issue3965: avoid a crash on Windows when filename is unicode
|
|
for name in (TESTFN, unicode(TESTFN), unicode(TESTFN + '\t')):
|
|
try:
|
|
f = open(name, "rr")
|
|
except (IOError, ValueError):
|
|
pass
|
|
else:
|
|
f.close()
|
|
|
|
def testStdinSeek(self):
|
|
if sys.platform == 'osf1V5':
|
|
# This causes the interpreter to exit on OSF1 v5.1.
|
|
self.skipTest('Skipping sys.stdin.seek(-1), it may crash '
|
|
'the interpreter. Test manually.')
|
|
|
|
if not sys.stdin.isatty():
|
|
# Issue #23168: if stdin is redirected to a file, stdin becomes
|
|
# seekable
|
|
self.skipTest('stdin must be a TTY in this test')
|
|
|
|
self.assertRaises(IOError, sys.stdin.seek, -1)
|
|
|
|
def testStdinTruncate(self):
|
|
self.assertRaises(IOError, sys.stdin.truncate)
|
|
|
|
def testUnicodeOpen(self):
|
|
# verify repr works for unicode too
|
|
f = open(unicode(TESTFN), "w")
|
|
self.assertTrue(repr(f).startswith("<open file u'" + TESTFN))
|
|
f.close()
|
|
os.unlink(TESTFN)
|
|
|
|
def testBadModeArgument(self):
|
|
# verify that we get a sensible error message for bad mode argument
|
|
bad_mode = "qwerty"
|
|
try:
|
|
f = open(TESTFN, bad_mode)
|
|
except ValueError, msg:
|
|
if msg.args[0] != 0:
|
|
s = str(msg)
|
|
if TESTFN in s or bad_mode not in s:
|
|
self.fail("bad error message for invalid mode: %s" % s)
|
|
# if msg.args[0] == 0, we're probably on Windows where there may
|
|
# be no obvious way to discover why open() failed.
|
|
else:
|
|
f.close()
|
|
self.fail("no error for invalid mode: %s" % bad_mode)
|
|
|
|
def testSetBufferSize(self):
|
|
# make sure that explicitly setting the buffer size doesn't cause
|
|
# misbehaviour especially with repeated close() calls
|
|
for s in (-1, 0, 1, 512):
|
|
try:
|
|
f = open(TESTFN, 'w', s)
|
|
f.write(str(s))
|
|
f.close()
|
|
f.close()
|
|
f = open(TESTFN, 'r', s)
|
|
d = int(f.read())
|
|
f.close()
|
|
f.close()
|
|
except IOError, msg:
|
|
self.fail('error setting buffer size %d: %s' % (s, str(msg)))
|
|
self.assertEqual(d, s)
|
|
|
|
def testTruncateOnWindows(self):
|
|
os.unlink(TESTFN)
|
|
|
|
def bug801631():
|
|
# SF bug <http://www.python.org/sf/801631>
|
|
# "file.truncate fault on windows"
|
|
f = open(TESTFN, 'wb')
|
|
f.write('12345678901') # 11 bytes
|
|
f.close()
|
|
|
|
f = open(TESTFN,'rb+')
|
|
data = f.read(5)
|
|
if data != '12345':
|
|
self.fail("Read on file opened for update failed %r" % data)
|
|
if f.tell() != 5:
|
|
self.fail("File pos after read wrong %d" % f.tell())
|
|
|
|
f.truncate()
|
|
if f.tell() != 5:
|
|
self.fail("File pos after ftruncate wrong %d" % f.tell())
|
|
|
|
f.close()
|
|
size = os.path.getsize(TESTFN)
|
|
if size != 5:
|
|
self.fail("File size after ftruncate wrong %d" % size)
|
|
|
|
try:
|
|
bug801631()
|
|
finally:
|
|
os.unlink(TESTFN)
|
|
|
|
def testIteration(self):
|
|
# Test the complex interaction when mixing file-iteration and the
|
|
# various read* methods. Ostensibly, the mixture could just be tested
|
|
# to work when it should work according to the Python language,
|
|
# instead of fail when it should fail according to the current CPython
|
|
# implementation. People don't always program Python the way they
|
|
# should, though, and the implemenation might change in subtle ways,
|
|
# so we explicitly test for errors, too; the test will just have to
|
|
# be updated when the implementation changes.
|
|
dataoffset = 16384
|
|
filler = "ham\n"
|
|
assert not dataoffset % len(filler), \
|
|
"dataoffset must be multiple of len(filler)"
|
|
nchunks = dataoffset // len(filler)
|
|
testlines = [
|
|
"spam, spam and eggs\n",
|
|
"eggs, spam, ham and spam\n",
|
|
"saussages, spam, spam and eggs\n",
|
|
"spam, ham, spam and eggs\n",
|
|
"spam, spam, spam, spam, spam, ham, spam\n",
|
|
"wonderful spaaaaaam.\n"
|
|
]
|
|
methods = [("readline", ()), ("read", ()), ("readlines", ()),
|
|
("readinto", (array("c", " "*100),))]
|
|
|
|
try:
|
|
# Prepare the testfile
|
|
bag = open(TESTFN, "w")
|
|
bag.write(filler * nchunks)
|
|
bag.writelines(testlines)
|
|
bag.close()
|
|
# Test for appropriate errors mixing read* and iteration
|
|
for methodname, args in methods:
|
|
f = open(TESTFN)
|
|
if f.next() != filler:
|
|
self.fail, "Broken testfile"
|
|
meth = getattr(f, methodname)
|
|
try:
|
|
meth(*args)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
self.fail("%s%r after next() didn't raise ValueError" %
|
|
(methodname, args))
|
|
f.close()
|
|
|
|
# Test to see if harmless (by accident) mixing of read* and
|
|
# iteration still works. This depends on the size of the internal
|
|
# iteration buffer (currently 8192,) but we can test it in a
|
|
# flexible manner. Each line in the bag o' ham is 4 bytes
|
|
# ("h", "a", "m", "\n"), so 4096 lines of that should get us
|
|
# exactly on the buffer boundary for any power-of-2 buffersize
|
|
# between 4 and 16384 (inclusive).
|
|
f = open(TESTFN)
|
|
for i in range(nchunks):
|
|
f.next()
|
|
testline = testlines.pop(0)
|
|
try:
|
|
line = f.readline()
|
|
except ValueError:
|
|
self.fail("readline() after next() with supposedly empty "
|
|
"iteration-buffer failed anyway")
|
|
if line != testline:
|
|
self.fail("readline() after next() with empty buffer "
|
|
"failed. Got %r, expected %r" % (line, testline))
|
|
testline = testlines.pop(0)
|
|
buf = array("c", "\x00" * len(testline))
|
|
try:
|
|
f.readinto(buf)
|
|
except ValueError:
|
|
self.fail("readinto() after next() with supposedly empty "
|
|
"iteration-buffer failed anyway")
|
|
line = buf.tostring()
|
|
if line != testline:
|
|
self.fail("readinto() after next() with empty buffer "
|
|
"failed. Got %r, expected %r" % (line, testline))
|
|
|
|
testline = testlines.pop(0)
|
|
try:
|
|
line = f.read(len(testline))
|
|
except ValueError:
|
|
self.fail("read() after next() with supposedly empty "
|
|
"iteration-buffer failed anyway")
|
|
if line != testline:
|
|
self.fail("read() after next() with empty buffer "
|
|
"failed. Got %r, expected %r" % (line, testline))
|
|
try:
|
|
lines = f.readlines()
|
|
except ValueError:
|
|
self.fail("readlines() after next() with supposedly empty "
|
|
"iteration-buffer failed anyway")
|
|
if lines != testlines:
|
|
self.fail("readlines() after next() with empty buffer "
|
|
"failed. Got %r, expected %r" % (line, testline))
|
|
# Reading after iteration hit EOF shouldn't hurt either
|
|
f = open(TESTFN)
|
|
try:
|
|
for line in f:
|
|
pass
|
|
try:
|
|
f.readline()
|
|
f.readinto(buf)
|
|
f.read()
|
|
f.readlines()
|
|
except ValueError:
|
|
self.fail("read* failed after next() consumed file")
|
|
finally:
|
|
f.close()
|
|
finally:
|
|
os.unlink(TESTFN)
|
|
|
|
@unittest.skipUnless(os.name == 'posix', 'test requires a posix system.')
|
|
def test_write_full(self):
|
|
devfull = '/dev/full'
|
|
if not (os.path.exists(devfull) and
|
|
stat.S_ISCHR(os.stat(devfull).st_mode)):
|
|
# Issue #21934: OpenBSD does not have a /dev/full character device
|
|
self.skipTest('requires %r' % devfull)
|
|
with open(devfull, 'wb', 1) as f:
|
|
with self.assertRaises(IOError):
|
|
f.write('hello\n')
|
|
with open(devfull, 'wb', 1) as f:
|
|
with self.assertRaises(IOError):
|
|
# Issue #17976
|
|
f.write('hello')
|
|
f.write('\n')
|
|
with open(devfull, 'wb', 0) as f:
|
|
with self.assertRaises(IOError):
|
|
f.write('h')
|
|
|
|
@unittest.skipUnless(sys.maxsize > 2**31, "requires 64-bit system")
|
|
@test_support.precisionbigmemtest(2**31, 2.5, dry_run=False)
|
|
def test_very_long_line(self, size):
|
|
# Issue #22526
|
|
requires('largefile')
|
|
with open(TESTFN, "wb") as fp:
|
|
fp.seek(size - 1)
|
|
fp.write("\0")
|
|
with open(TESTFN, "rb") as fp:
|
|
for l in fp:
|
|
pass
|
|
self.assertEqual(len(l), size)
|
|
self.assertEqual(l.count("\0"), size)
|
|
l = None
|
|
|
|
class FileSubclassTests(unittest.TestCase):
|
|
|
|
def testExit(self):
|
|
# test that exiting with context calls subclass' close
|
|
class C(file):
|
|
def __init__(self, *args):
|
|
self.subclass_closed = False
|
|
file.__init__(self, *args)
|
|
def close(self):
|
|
self.subclass_closed = True
|
|
file.close(self)
|
|
|
|
with C(TESTFN, 'w') as f:
|
|
pass
|
|
self.assertTrue(f.subclass_closed)
|
|
|
|
|
|
@unittest.skipUnless(threading, 'Threading required for this test.')
|
|
class FileThreadingTests(unittest.TestCase):
|
|
# These tests check the ability to call various methods of file objects
|
|
# (including close()) concurrently without crashing the Python interpreter.
|
|
# See #815646, #595601
|
|
|
|
def setUp(self):
|
|
self._threads = test_support.threading_setup()
|
|
self.f = None
|
|
self.filename = TESTFN
|
|
with open(self.filename, "w") as f:
|
|
f.write("\n".join("0123456789"))
|
|
self._count_lock = threading.Lock()
|
|
self.close_count = 0
|
|
self.close_success_count = 0
|
|
self.use_buffering = False
|
|
|
|
def tearDown(self):
|
|
if self.f:
|
|
try:
|
|
self.f.close()
|
|
except (EnvironmentError, ValueError):
|
|
pass
|
|
try:
|
|
os.remove(self.filename)
|
|
except EnvironmentError:
|
|
pass
|
|
test_support.threading_cleanup(*self._threads)
|
|
|
|
def _create_file(self):
|
|
if self.use_buffering:
|
|
self.f = open(self.filename, "w+", buffering=1024*16)
|
|
else:
|
|
self.f = open(self.filename, "w+")
|
|
|
|
def _close_file(self):
|
|
with self._count_lock:
|
|
self.close_count += 1
|
|
self.f.close()
|
|
with self._count_lock:
|
|
self.close_success_count += 1
|
|
|
|
def _close_and_reopen_file(self):
|
|
self._close_file()
|
|
# if close raises an exception thats fine, self.f remains valid so
|
|
# we don't need to reopen.
|
|
self._create_file()
|
|
|
|
def _run_workers(self, func, nb_workers, duration=0.2):
|
|
with self._count_lock:
|
|
self.close_count = 0
|
|
self.close_success_count = 0
|
|
self.do_continue = True
|
|
threads = []
|
|
try:
|
|
for i in range(nb_workers):
|
|
t = threading.Thread(target=func)
|
|
t.start()
|
|
threads.append(t)
|
|
for _ in xrange(100):
|
|
time.sleep(duration/100)
|
|
with self._count_lock:
|
|
if self.close_count-self.close_success_count > nb_workers+1:
|
|
if test_support.verbose:
|
|
print 'Q',
|
|
break
|
|
time.sleep(duration)
|
|
finally:
|
|
self.do_continue = False
|
|
for t in threads:
|
|
t.join()
|
|
|
|
def _test_close_open_io(self, io_func, nb_workers=5):
|
|
def worker():
|
|
self._create_file()
|
|
funcs = itertools.cycle((
|
|
lambda: io_func(),
|
|
lambda: self._close_and_reopen_file(),
|
|
))
|
|
for f in funcs:
|
|
if not self.do_continue:
|
|
break
|
|
try:
|
|
f()
|
|
except (IOError, ValueError):
|
|
pass
|
|
self._run_workers(worker, nb_workers)
|
|
if test_support.verbose:
|
|
# Useful verbose statistics when tuning this test to take
|
|
# less time to run but still ensuring that its still useful.
|
|
#
|
|
# the percent of close calls that raised an error
|
|
percent = 100. - 100.*self.close_success_count/self.close_count
|
|
print self.close_count, ('%.4f ' % percent),
|
|
|
|
def test_close_open(self):
|
|
def io_func():
|
|
pass
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_flush(self):
|
|
def io_func():
|
|
self.f.flush()
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_iter(self):
|
|
def io_func():
|
|
list(iter(self.f))
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_isatty(self):
|
|
def io_func():
|
|
self.f.isatty()
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_print(self):
|
|
def io_func():
|
|
print >> self.f, ''
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_print_buffered(self):
|
|
self.use_buffering = True
|
|
def io_func():
|
|
print >> self.f, ''
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_read(self):
|
|
def io_func():
|
|
self.f.read(0)
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_readinto(self):
|
|
def io_func():
|
|
a = array('c', 'xxxxx')
|
|
self.f.readinto(a)
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_readline(self):
|
|
def io_func():
|
|
self.f.readline()
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_readlines(self):
|
|
def io_func():
|
|
self.f.readlines()
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_seek(self):
|
|
def io_func():
|
|
self.f.seek(0, 0)
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_tell(self):
|
|
def io_func():
|
|
self.f.tell()
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_truncate(self):
|
|
def io_func():
|
|
self.f.truncate()
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_write(self):
|
|
def io_func():
|
|
self.f.write('')
|
|
self._test_close_open_io(io_func)
|
|
|
|
def test_close_open_writelines(self):
|
|
def io_func():
|
|
self.f.writelines('')
|
|
self._test_close_open_io(io_func)
|
|
|
|
|
|
@unittest.skipUnless(os.name == 'posix', 'test requires a posix system.')
|
|
class TestFileSignalEINTR(unittest.TestCase):
|
|
def _test_reading(self, data_to_write, read_and_verify_code, method_name,
|
|
universal_newlines=False):
|
|
"""Generic buffered read method test harness to verify EINTR behavior.
|
|
|
|
Also validates that Python signal handlers are run during the read.
|
|
|
|
Args:
|
|
data_to_write: String to write to the child process for reading
|
|
before sending it a signal, confirming the signal was handled,
|
|
writing a final newline char and closing the infile pipe.
|
|
read_and_verify_code: Single "line" of code to read from a file
|
|
object named 'infile' and validate the result. This will be
|
|
executed as part of a python subprocess fed data_to_write.
|
|
method_name: The name of the read method being tested, for use in
|
|
an error message on failure.
|
|
universal_newlines: If True, infile will be opened in universal
|
|
newline mode in the child process.
|
|
"""
|
|
if universal_newlines:
|
|
# Test the \r\n -> \n conversion while we're at it.
|
|
data_to_write = data_to_write.replace('\n', '\r\n')
|
|
infile_setup_code = 'infile = os.fdopen(sys.stdin.fileno(), "rU")'
|
|
else:
|
|
infile_setup_code = 'infile = sys.stdin'
|
|
# Total pipe IO in this function is smaller than the minimum posix OS
|
|
# pipe buffer size of 512 bytes. No writer should block.
|
|
assert len(data_to_write) < 512, 'data_to_write must fit in pipe buf.'
|
|
|
|
child_code = (
|
|
'import os, signal, sys ;'
|
|
'signal.signal('
|
|
'signal.SIGINT, lambda s, f: sys.stderr.write("$\\n")) ;'
|
|
+ infile_setup_code + ' ;' +
|
|
'assert isinstance(infile, file) ;'
|
|
'sys.stderr.write("Go.\\n") ;'
|
|
+ read_and_verify_code)
|
|
reader_process = subprocess.Popen(
|
|
[sys.executable, '-c', child_code],
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
# Wait for the signal handler to be installed.
|
|
go = reader_process.stderr.read(4)
|
|
if go != 'Go.\n':
|
|
reader_process.kill()
|
|
self.fail('Error from %s process while awaiting "Go":\n%s' % (
|
|
method_name, go+reader_process.stderr.read()))
|
|
reader_process.stdin.write(data_to_write)
|
|
signals_sent = 0
|
|
rlist = []
|
|
# We don't know when the read_and_verify_code in our child is actually
|
|
# executing within the read system call we want to interrupt. This
|
|
# loop waits for a bit before sending the first signal to increase
|
|
# the likelihood of that. Implementations without correct EINTR
|
|
# and signal handling usually fail this test.
|
|
while not rlist:
|
|
rlist, _, _ = select.select([reader_process.stderr], (), (), 0.05)
|
|
reader_process.send_signal(signal.SIGINT)
|
|
# Give the subprocess time to handle it before we loop around and
|
|
# send another one. On OSX the second signal happening close to
|
|
# immediately after the first was causing the subprocess to crash
|
|
# via the OS's default SIGINT handler.
|
|
time.sleep(0.1)
|
|
signals_sent += 1
|
|
if signals_sent > 200:
|
|
reader_process.kill()
|
|
self.fail("failed to handle signal during %s." % method_name)
|
|
# This assumes anything unexpected that writes to stderr will also
|
|
# write a newline. That is true of the traceback printing code.
|
|
signal_line = reader_process.stderr.readline()
|
|
if signal_line != '$\n':
|
|
reader_process.kill()
|
|
self.fail('Error from %s process while awaiting signal:\n%s' % (
|
|
method_name, signal_line+reader_process.stderr.read()))
|
|
# We append a newline to our input so that a readline call can
|
|
# end on its own before the EOF is seen.
|
|
stdout, stderr = reader_process.communicate(input='\n')
|
|
if reader_process.returncode != 0:
|
|
self.fail('%s() process exited rc=%d.\nSTDOUT:\n%s\nSTDERR:\n%s' % (
|
|
method_name, reader_process.returncode, stdout, stderr))
|
|
|
|
def test_readline(self, universal_newlines=False):
|
|
"""file.readline must handle signals and not lose data."""
|
|
self._test_reading(
|
|
data_to_write='hello, world!',
|
|
read_and_verify_code=(
|
|
'line = infile.readline() ;'
|
|
'expected_line = "hello, world!\\n" ;'
|
|
'assert line == expected_line, ('
|
|
'"read %r expected %r" % (line, expected_line))'
|
|
),
|
|
method_name='readline',
|
|
universal_newlines=universal_newlines)
|
|
|
|
def test_readline_with_universal_newlines(self):
|
|
self.test_readline(universal_newlines=True)
|
|
|
|
def test_readlines(self, universal_newlines=False):
|
|
"""file.readlines must handle signals and not lose data."""
|
|
self._test_reading(
|
|
data_to_write='hello\nworld!',
|
|
read_and_verify_code=(
|
|
'lines = infile.readlines() ;'
|
|
'expected_lines = ["hello\\n", "world!\\n"] ;'
|
|
'assert lines == expected_lines, ('
|
|
'"readlines returned wrong data.\\n" '
|
|
'"got lines %r\\nexpected %r" '
|
|
'% (lines, expected_lines))'
|
|
),
|
|
method_name='readlines',
|
|
universal_newlines=universal_newlines)
|
|
|
|
def test_readlines_with_universal_newlines(self):
|
|
self.test_readlines(universal_newlines=True)
|
|
|
|
def test_readall(self):
|
|
"""Unbounded file.read() must handle signals and not lose data."""
|
|
self._test_reading(
|
|
data_to_write='hello, world!abcdefghijklm',
|
|
read_and_verify_code=(
|
|
'data = infile.read() ;'
|
|
'expected_data = "hello, world!abcdefghijklm\\n";'
|
|
'assert data == expected_data, ('
|
|
'"read %r expected %r" % (data, expected_data))'
|
|
),
|
|
method_name='unbounded read')
|
|
|
|
def test_readinto(self):
|
|
"""file.readinto must handle signals and not lose data."""
|
|
self._test_reading(
|
|
data_to_write='hello, world!',
|
|
read_and_verify_code=(
|
|
'data = bytearray(50) ;'
|
|
'num_read = infile.readinto(data) ;'
|
|
'expected_data = "hello, world!\\n";'
|
|
'assert data[:num_read] == expected_data, ('
|
|
'"read %r expected %r" % (data, expected_data))'
|
|
),
|
|
method_name='readinto')
|
|
|
|
|
|
class StdoutTests(unittest.TestCase):
|
|
|
|
def test_move_stdout_on_write(self):
|
|
# Issue 3242: sys.stdout can be replaced (and freed) during a
|
|
# print statement; prevent a segfault in this case
|
|
save_stdout = sys.stdout
|
|
|
|
class File:
|
|
def write(self, data):
|
|
if '\n' in data:
|
|
sys.stdout = save_stdout
|
|
|
|
try:
|
|
sys.stdout = File()
|
|
print "some text"
|
|
finally:
|
|
sys.stdout = save_stdout
|
|
|
|
def test_del_stdout_before_print(self):
|
|
# Issue 4597: 'print' with no argument wasn't reporting when
|
|
# sys.stdout was deleted.
|
|
save_stdout = sys.stdout
|
|
del sys.stdout
|
|
try:
|
|
print
|
|
except RuntimeError as e:
|
|
self.assertEqual(str(e), "lost sys.stdout")
|
|
else:
|
|
self.fail("Expected RuntimeError")
|
|
finally:
|
|
sys.stdout = save_stdout
|
|
|
|
def test_unicode(self):
|
|
import subprocess
|
|
|
|
def get_message(encoding, *code):
|
|
code = '\n'.join(code)
|
|
env = os.environ.copy()
|
|
env['PYTHONIOENCODING'] = encoding
|
|
process = subprocess.Popen([sys.executable, "-c", code],
|
|
stdout=subprocess.PIPE, env=env)
|
|
stdout, stderr = process.communicate()
|
|
self.assertEqual(process.returncode, 0)
|
|
return stdout
|
|
|
|
def check_message(text, encoding, expected):
|
|
stdout = get_message(encoding,
|
|
"import sys",
|
|
"sys.stdout.write(%r)" % text,
|
|
"sys.stdout.flush()")
|
|
self.assertEqual(stdout, expected)
|
|
|
|
# test the encoding
|
|
check_message(u'15\u20ac', "iso-8859-15", "15\xa4")
|
|
check_message(u'15\u20ac', "utf-8", '15\xe2\x82\xac')
|
|
check_message(u'15\u20ac', "utf-16-le", '1\x005\x00\xac\x20')
|
|
|
|
# test the error handler
|
|
check_message(u'15\u20ac', "iso-8859-1:ignore", "15")
|
|
check_message(u'15\u20ac', "iso-8859-1:replace", "15?")
|
|
check_message(u'15\u20ac', "iso-8859-1:backslashreplace", "15\\u20ac")
|
|
|
|
# test the buffer API
|
|
for objtype in ('buffer', 'bytearray'):
|
|
stdout = get_message('ascii',
|
|
'import sys',
|
|
r'sys.stdout.write(%s("\xe9"))' % objtype,
|
|
'sys.stdout.flush()')
|
|
self.assertEqual(stdout, "\xe9")
|
|
|
|
|
|
def test_main():
|
|
# Historically, these tests have been sloppy about removing TESTFN.
|
|
# So get rid of it no matter what.
|
|
try:
|
|
run_unittest(AutoFileTests, OtherFileTests, FileSubclassTests,
|
|
FileThreadingTests, TestFileSignalEINTR, StdoutTests)
|
|
finally:
|
|
if os.path.exists(TESTFN):
|
|
os.unlink(TESTFN)
|
|
|
|
if __name__ == '__main__':
|
|
test_main()
|