|
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) |
|
|
|
def test_iteration_torture(self): |
|
# bpo-31530 |
|
with open(self.filename, "wb") as fp: |
|
for i in xrange(2**20): |
|
fp.write(b"0"*50 + b"\n") |
|
with open(self.filename, "rb") as f: |
|
def it(): |
|
for l in f: |
|
pass |
|
self._run_workers(it, 10) |
|
|
|
def test_iteration_seek(self): |
|
# bpo-31530: Crash when concurrently seek and iterate over a file. |
|
with open(self.filename, "wb") as fp: |
|
for i in xrange(10000): |
|
fp.write(b"0"*50 + b"\n") |
|
with open(self.filename, "rb") as f: |
|
it = iter([1] + [0]*10) # one thread reads, others seek |
|
def iterate(): |
|
if next(it): |
|
for l in f: |
|
pass |
|
else: |
|
for i in xrange(100): |
|
f.seek(i*100, 0) |
|
self._run_workers(iterate, 10) |
|
|
|
|
|
@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() |
