Try to restore the old test_file and test_univnewlines as new, different files
(with the right revisions this time, hopefully)
diff --git a/Lib/test/test_file.py b/Lib/test/test_file.py
index 4b0c759..1974f56 100644
--- a/Lib/test/test_file.py
+++ b/Lib/test/test_file.py
@@ -1,3 +1,6 @@
+# NOTE: this file tests the new `io` library backported from Python 3.x.
+# Similar tests for the builtin file object can be found in test_file2k.py.
+
from __future__ import print_function
import sys
diff --git a/Lib/test/test_file2k.py b/Lib/test/test_file2k.py
index 4b0c759..a134a89 100644
--- a/Lib/test/test_file2k.py
+++ b/Lib/test/test_file2k.py
@@ -1,14 +1,13 @@
-from __future__ import print_function
-
import sys
import os
import unittest
+import itertools
+import time
+import threading
from array import array
from weakref import proxy
-import io
-import _pyio as pyio
-
+from test import test_support
from test.test_support import TESTFN, findfile, run_unittest
from UserList import UserList
@@ -16,7 +15,7 @@
# file tests for which a test file is automatically set up
def setUp(self):
- self.f = self.open(TESTFN, 'wb')
+ self.f = open(TESTFN, 'wb')
def tearDown(self):
if self.f:
@@ -26,7 +25,7 @@
def testWeakRefs(self):
# verify weak references
p = proxy(self.f)
- p.write(b'teststring')
+ p.write('teststring')
self.assertEquals(self.f.tell(), p.tell())
self.f.close()
self.f = None
@@ -35,35 +34,35 @@
def testAttributes(self):
# verify expected attributes exist
f = self.f
+ softspace = f.softspace
f.name # merely shouldn't blow up
f.mode # ditto
f.closed # ditto
+ # 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(b'12')
+ self.f.write('12')
self.f.close()
- a = array('b', b'x'*10)
- self.f = self.open(TESTFN, 'rb')
+ a = array('c', 'x'*10)
+ self.f = open(TESTFN, 'rb')
n = self.f.readinto(a)
- self.assertEquals(b'12', a.tostring()[:n])
-
- def testReadinto_text(self):
- # verify readinto refuses text files
- a = array('b', b'x'*10)
- self.f.close()
- self.f = self.open(TESTFN, 'r')
- if hasattr(self.f, "readinto"):
- self.assertRaises(TypeError, self.f.readinto, a)
+ self.assertEquals('12', a.tostring()[:n])
def testWritelinesUserList(self):
# verify writelines with instance sequence
- l = UserList([b'1', b'2'])
+ l = UserList(['1', '2'])
self.f.writelines(l)
self.f.close()
- self.f = self.open(TESTFN, 'rb')
+ self.f = open(TESTFN, 'rb')
buf = self.f.read()
- self.assertEquals(buf, b'12')
+ self.assertEquals(buf, '12')
def testWritelinesIntegers(self):
# verify writelines with integers
@@ -82,43 +81,36 @@
self.assertRaises(TypeError, self.f.writelines,
[NonString(), NonString()])
+ def testRepr(self):
+ # verify repr works
+ self.assert_(repr(self.f).startswith("<open file '" + TESTFN))
+
def testErrors(self):
f = self.f
self.assertEquals(f.name, TESTFN)
self.assert_(not f.isatty())
self.assert_(not f.closed)
- if hasattr(f, "readinto"):
- self.assertRaises((IOError, TypeError), f.readinto, "")
+ self.assertRaises(TypeError, f.readinto, "")
f.close()
self.assert_(f.closed)
def testMethods(self):
- methods = [('fileno', ()),
- ('flush', ()),
- ('isatty', ()),
- ('next', ()),
- ('read', ()),
- ('write', (b"",)),
- ('readline', ()),
- ('readlines', ()),
- ('seek', (0,)),
- ('tell', ()),
- ('write', (b"",)),
- ('writelines', ([],)),
- ('__iter__', ()),
- ]
- if not sys.platform.startswith('atheos'):
- methods.append(('truncate', ()))
+ methods = ['fileno', 'flush', 'isatty', 'next', 'read', 'readinto',
+ 'readline', 'readlines', 'seek', 'tell', 'truncate',
+ 'write', 'xreadlines', '__iter__']
+ if sys.platform.startswith('atheos'):
+ methods.remove('truncate')
# __exit__ should close the file
self.f.__exit__(None, None, None)
self.assert_(self.f.closed)
- for methodname, args in methods:
+ for methodname in methods:
method = getattr(self.f, methodname)
# should raise on closed file
- self.assertRaises(ValueError, method, *args)
+ self.assertRaises(ValueError, method)
+ self.assertRaises(ValueError, self.f.writelines, [])
# file is closed, __exit__ shouldn't do anything
self.assertEquals(self.f.__exit__(None, None, None), None)
@@ -131,47 +123,70 @@
def testReadWhenWriting(self):
self.assertRaises(IOError, self.f.read)
-class CAutoFileTests(AutoFileTests):
- open = io.open
-
-class PyAutoFileTests(AutoFileTests):
- open = staticmethod(pyio.open)
-
-
class OtherFileTests(unittest.TestCase):
+ def testOpenDir(self):
+ this_dir = os.path.dirname(__file__)
+ 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 = self.open(TESTFN, mode)
+ 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 testStdin(self):
# This causes the interpreter to exit on OSF1 v5.1.
if sys.platform != 'osf1V5':
- self.assertRaises((IOError, ValueError), sys.stdin.seek, -1)
+ self.assertRaises(IOError, sys.stdin.seek, -1)
else:
- print((
+ print >>sys.__stdout__, (
' Skipping sys.stdin.seek(-1), it may crash the interpreter.'
- ' Test manually.'), file=sys.__stdout__)
- self.assertRaises((IOError, ValueError), sys.stdin.truncate)
+ ' Test manually.')
+ self.assertRaises(IOError, sys.stdin.truncate)
+
+ def testUnicodeOpen(self):
+ # verify repr works for unicode too
+ f = open(unicode(TESTFN), "w")
+ self.assert_(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 = self.open(TESTFN, bad_mode)
- except ValueError as msg:
- if msg.args[0] != 0:
+ f = open(TESTFN, bad_mode)
+ except ValueError, msg:
+ if msg[0] != 0:
s = str(msg)
if s.find(TESTFN) != -1 or s.find(bad_mode) == -1:
self.fail("bad error message for invalid mode: %s" % s)
- # if msg.args[0] == 0, we're probably on Windows where there may be
+ # if msg[0] == 0, we're probably on Windows where there may be
# no obvious way to discover why open() failed.
else:
f.close()
@@ -182,32 +197,31 @@
# misbehaviour especially with repeated close() calls
for s in (-1, 0, 1, 512):
try:
- f = self.open(TESTFN, 'wb', s)
- f.write(str(s).encode("ascii"))
+ f = open(TESTFN, 'w', s)
+ f.write(str(s))
f.close()
f.close()
- f = self.open(TESTFN, 'rb', s)
- d = int(f.read().decode("ascii"))
+ f = open(TESTFN, 'r', s)
+ d = int(f.read())
f.close()
f.close()
- except IOError as msg:
+ except IOError, msg:
self.fail('error setting buffer size %d: %s' % (s, str(msg)))
self.assertEquals(d, s)
def testTruncateOnWindows(self):
- # SF bug <http://www.python.org/sf/801631>
- # "file.truncate fault on windows"
-
os.unlink(TESTFN)
- f = self.open(TESTFN, 'wb')
- try:
- f.write(b'12345678901') # 11 bytes
+ 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 = self.open(TESTFN,'rb+')
+ f = open(TESTFN,'rb+')
data = f.read(5)
- if data != b'12345':
+ 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())
@@ -220,42 +234,56 @@
size = os.path.getsize(TESTFN)
if size != 5:
self.fail("File size after ftruncate wrong %d" % size)
+
+ try:
+ bug801631()
finally:
- f.close()
os.unlink(TESTFN)
def testIteration(self):
# Test the complex interaction when mixing file-iteration and the
- # various read* methods.
+ # 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 = b"ham\n"
+ filler = "ham\n"
assert not dataoffset % len(filler), \
"dataoffset must be multiple of len(filler)"
nchunks = dataoffset // len(filler)
testlines = [
- b"spam, spam and eggs\n",
- b"eggs, spam, ham and spam\n",
- b"saussages, spam, spam and eggs\n",
- b"spam, ham, spam and eggs\n",
- b"spam, spam, spam, spam, spam, ham, spam\n",
- b"wonderful spaaaaaam.\n"
+ "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("b", b" "*100),))]
+ ("readinto", (array("c", " "*100),))]
try:
# Prepare the testfile
- bag = self.open(TESTFN, "wb")
+ 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 = self.open(TESTFN, 'rb')
- if next(f) != filler:
+ f = open(TESTFN)
+ if f.next() != filler:
self.fail, "Broken testfile"
meth = getattr(f, methodname)
- meth(*args) # This simply shouldn't fail
+ 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
@@ -265,9 +293,9 @@
# ("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 = self.open(TESTFN, 'rb')
+ f = open(TESTFN)
for i in range(nchunks):
- next(f)
+ f.next()
testline = testlines.pop(0)
try:
line = f.readline()
@@ -278,7 +306,7 @@
self.fail("readline() after next() with empty buffer "
"failed. Got %r, expected %r" % (line, testline))
testline = testlines.pop(0)
- buf = array("b", b"\x00" * len(testline))
+ buf = array("c", "\x00" * len(testline))
try:
f.readinto(buf)
except ValueError:
@@ -307,7 +335,7 @@
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 = self.open(TESTFN, 'rb')
+ f = open(TESTFN)
try:
for line in f:
pass
@@ -323,19 +351,222 @@
finally:
os.unlink(TESTFN)
-class COtherFileTests(OtherFileTests):
- open = io.open
+class FileSubclassTests(unittest.TestCase):
-class PyOtherFileTests(OtherFileTests):
- open = staticmethod(pyio.open)
+ 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.failUnless(f.subclass_closed)
+
+
+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.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
+
+ def tearDown(self):
+ if self.f:
+ try:
+ self.f.close()
+ except (EnvironmentError, ValueError):
+ pass
+ try:
+ os.remove(self.filename)
+ except EnvironmentError:
+ pass
+
+ def _create_file(self):
+ 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_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)
+
+
+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.assertEquals(str(e), "lost sys.stdout")
+ else:
+ self.fail("Expected RuntimeError")
+ finally:
+ sys.stdout = save_stdout
def test_main():
# Historically, these tests have been sloppy about removing TESTFN.
# So get rid of it no matter what.
try:
- run_unittest(CAutoFileTests, PyAutoFileTests,
- COtherFileTests, PyOtherFileTests)
+ run_unittest(AutoFileTests, OtherFileTests, FileSubclassTests,
+ FileThreadingTests, StdoutTests)
finally:
if os.path.exists(TESTFN):
os.unlink(TESTFN)
diff --git a/Lib/test/test_univnewlines.py b/Lib/test/test_univnewlines.py
index 1f7352a..a0a9090 100644
--- a/Lib/test/test_univnewlines.py
+++ b/Lib/test/test_univnewlines.py
@@ -1,5 +1,8 @@
# Tests universal newline support for both reading and parsing files.
+# NOTE: this file tests the new `io` library backported from Python 3.x.
+# Similar tests for the builtin file object can be found in test_univnewlines2k.py.
+
from __future__ import print_function
from __future__ import unicode_literals
diff --git a/Lib/test/test_univnewlines2k.py b/Lib/test/test_univnewlines2k.py
new file mode 100644
index 0000000..63c6fe8
--- /dev/null
+++ b/Lib/test/test_univnewlines2k.py
@@ -0,0 +1,129 @@
+# Tests universal newline support for both reading and parsing files.
+import unittest
+import os
+import sys
+from test import test_support
+
+if not hasattr(sys.stdin, 'newlines'):
+ raise unittest.SkipTest, \
+ "This Python does not have universal newline support"
+
+FATX = 'x' * (2**14)
+
+DATA_TEMPLATE = [
+ "line1=1",
+ "line2='this is a very long line designed to go past the magic " +
+ "hundred character limit that is inside fileobject.c and which " +
+ "is meant to speed up the common case, but we also want to test " +
+ "the uncommon case, naturally.'",
+ "def line3():pass",
+ "line4 = '%s'" % FATX,
+ ]
+
+DATA_LF = "\n".join(DATA_TEMPLATE) + "\n"
+DATA_CR = "\r".join(DATA_TEMPLATE) + "\r"
+DATA_CRLF = "\r\n".join(DATA_TEMPLATE) + "\r\n"
+
+# Note that DATA_MIXED also tests the ability to recognize a lone \r
+# before end-of-file.
+DATA_MIXED = "\n".join(DATA_TEMPLATE) + "\r"
+DATA_SPLIT = [x + "\n" for x in DATA_TEMPLATE]
+del x
+
+class TestGenericUnivNewlines(unittest.TestCase):
+ # use a class variable DATA to define the data to write to the file
+ # and a class variable NEWLINE to set the expected newlines value
+ READMODE = 'U'
+ WRITEMODE = 'wb'
+
+ def setUp(self):
+ with open(test_support.TESTFN, self.WRITEMODE) as fp:
+ fp.write(self.DATA)
+
+ def tearDown(self):
+ try:
+ os.unlink(test_support.TESTFN)
+ except:
+ pass
+
+ def test_read(self):
+ with open(test_support.TESTFN, self.READMODE) as fp:
+ data = fp.read()
+ self.assertEqual(data, DATA_LF)
+ self.assertEqual(repr(fp.newlines), repr(self.NEWLINE))
+
+ def test_readlines(self):
+ with open(test_support.TESTFN, self.READMODE) as fp:
+ data = fp.readlines()
+ self.assertEqual(data, DATA_SPLIT)
+ self.assertEqual(repr(fp.newlines), repr(self.NEWLINE))
+
+ def test_readline(self):
+ with open(test_support.TESTFN, self.READMODE) as fp:
+ data = []
+ d = fp.readline()
+ while d:
+ data.append(d)
+ d = fp.readline()
+ self.assertEqual(data, DATA_SPLIT)
+ self.assertEqual(repr(fp.newlines), repr(self.NEWLINE))
+
+ def test_seek(self):
+ with open(test_support.TESTFN, self.READMODE) as fp:
+ fp.readline()
+ pos = fp.tell()
+ data = fp.readlines()
+ self.assertEqual(data, DATA_SPLIT[1:])
+ fp.seek(pos)
+ data = fp.readlines()
+ self.assertEqual(data, DATA_SPLIT[1:])
+
+ def test_execfile(self):
+ namespace = {}
+ execfile(test_support.TESTFN, namespace)
+ func = namespace['line3']
+ self.assertEqual(func.func_code.co_firstlineno, 3)
+ self.assertEqual(namespace['line4'], FATX)
+
+
+class TestNativeNewlines(TestGenericUnivNewlines):
+ NEWLINE = None
+ DATA = DATA_LF
+ READMODE = 'r'
+ WRITEMODE = 'w'
+
+class TestCRNewlines(TestGenericUnivNewlines):
+ NEWLINE = '\r'
+ DATA = DATA_CR
+
+class TestLFNewlines(TestGenericUnivNewlines):
+ NEWLINE = '\n'
+ DATA = DATA_LF
+
+class TestCRLFNewlines(TestGenericUnivNewlines):
+ NEWLINE = '\r\n'
+ DATA = DATA_CRLF
+
+ def test_tell(self):
+ with open(test_support.TESTFN, self.READMODE) as fp:
+ self.assertEqual(repr(fp.newlines), repr(None))
+ data = fp.readline()
+ pos = fp.tell()
+ self.assertEqual(repr(fp.newlines), repr(self.NEWLINE))
+
+class TestMixedNewlines(TestGenericUnivNewlines):
+ NEWLINE = ('\r', '\n')
+ DATA = DATA_MIXED
+
+
+def test_main():
+ test_support.run_unittest(
+ TestNativeNewlines,
+ TestCRNewlines,
+ TestLFNewlines,
+ TestCRLFNewlines,
+ TestMixedNewlines
+ )
+
+if __name__ == '__main__':
+ test_main()