From 44316691796413e7473c7c6c681c84e1ce1b44e7 Mon Sep 17 00:00:00 2001 From: "antoine.pitrou" Date: Fri, 12 Jun 2009 20:41:52 +0000 Subject: [PATCH] Try to restore the old test_file and test_univnewlines as new, different files (with the right revisions this time, hopefully) git-svn-id: http://svn.python.org/projects/python/trunk@73396 6015fed2-1504-0410-9fe1-9d1591cc4771 --- Lib/test/test_file.py | 3 + Lib/test/test_file2k.py | 421 ++++++++++++++++----- Lib/test/test_univnewlines.py | 3 + ...test_univnewlines.py => test_univnewlines2k.py} | 81 ++-- 4 files changed, 371 insertions(+), 137 deletions(-) copy Lib/test/{test_univnewlines.py => test_univnewlines2k.py} (57%) diff --git a/Lib/test/test_file.py b/Lib/test/test_file.py index 4b0c75929d..1974f5601d 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 4b0c75929d..a134a89c05 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 @@ class AutoFileTests(unittest.TestCase): # 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 @@ class AutoFileTests(unittest.TestCase): 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 @@ class AutoFileTests(unittest.TestCase): 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 @@ class AutoFileTests(unittest.TestCase): self.assertRaises(TypeError, self.f.writelines, [NonString(), NonString()]) + def testRepr(self): + # verify repr works + self.assert_(repr(self.f).startswith(">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(" - # "file.truncate fault on windows" - os.unlink(TESTFN) - f = self.open(TESTFN, 'wb') - try: - f.write(b'12345678901') # 11 bytes + def bug801631(): + # SF bug + # "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 @@ class OtherFileTests(unittest.TestCase): 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 @@ class OtherFileTests(unittest.TestCase): # ("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 @@ class OtherFileTests(unittest.TestCase): 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 @@ class OtherFileTests(unittest.TestCase): 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 @@ class OtherFileTests(unittest.TestCase): 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 1f7352aa81..a0a909058c 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_univnewlines.py b/Lib/test/test_univnewlines2k.py similarity index 57% copy from Lib/test/test_univnewlines.py copy to Lib/test/test_univnewlines2k.py index 1f7352aa81..63c6fe8240 100644 --- a/Lib/test/test_univnewlines.py +++ b/Lib/test/test_univnewlines2k.py @@ -1,25 +1,20 @@ # Tests universal newline support for both reading and parsing files. - -from __future__ import print_function -from __future__ import unicode_literals - -import io -import _pyio as pyio import unittest import os import sys -from test import test_support as support +from test import test_support if not hasattr(sys.stdin, 'newlines'): - raise unittest.SkipTest( - "This Python does not have universal newline support") + 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 any default " + - "buffer limits that exist in io.py but we also want to test " + + "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, @@ -33,50 +28,48 @@ DATA_CRLF = "\r\n".join(DATA_TEMPLATE) + "\r\n" # 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 = 'r' + READMODE = 'U' WRITEMODE = 'wb' def setUp(self): - data = self.DATA - if "b" in self.WRITEMODE: - data = data.encode("ascii") - with self.open(support.TESTFN, self.WRITEMODE) as fp: - fp.write(data) + with open(test_support.TESTFN, self.WRITEMODE) as fp: + fp.write(self.DATA) def tearDown(self): try: - os.unlink(support.TESTFN) + os.unlink(test_support.TESTFN) except: pass def test_read(self): - with self.open(support.TESTFN, self.READMODE) as fp: + with open(test_support.TESTFN, self.READMODE) as fp: data = fp.read() self.assertEqual(data, DATA_LF) - self.assertEqual(set(fp.newlines), set(self.NEWLINE)) + self.assertEqual(repr(fp.newlines), repr(self.NEWLINE)) def test_readlines(self): - with self.open(support.TESTFN, self.READMODE) as fp: + with open(test_support.TESTFN, self.READMODE) as fp: data = fp.readlines() self.assertEqual(data, DATA_SPLIT) - self.assertEqual(set(fp.newlines), set(self.NEWLINE)) + self.assertEqual(repr(fp.newlines), repr(self.NEWLINE)) def test_readline(self): - with self.open(support.TESTFN, self.READMODE) as fp: + 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(set(fp.newlines), set(self.NEWLINE)) + self.assertEqual(repr(fp.newlines), repr(self.NEWLINE)) def test_seek(self): - with self.open(support.TESTFN, self.READMODE) as fp: + with open(test_support.TESTFN, self.READMODE) as fp: fp.readline() pos = fp.tell() data = fp.readlines() @@ -85,6 +78,19 @@ class TestGenericUnivNewlines(unittest.TestCase): 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' @@ -99,7 +105,7 @@ class TestCRLFNewlines(TestGenericUnivNewlines): DATA = DATA_CRLF def test_tell(self): - with self.open(support.TESTFN, self.READMODE) as fp: + with open(test_support.TESTFN, self.READMODE) as fp: self.assertEqual(repr(fp.newlines), repr(None)) data = fp.readline() pos = fp.tell() @@ -111,22 +117,13 @@ class TestMixedNewlines(TestGenericUnivNewlines): def test_main(): - base_tests = (TestCRNewlines, - TestLFNewlines, - TestCRLFNewlines, - TestMixedNewlines) - tests = [] - # Test the C and Python implementations. - for test in base_tests: - class CTest(test): - open = io.open - CTest.__name__ = str("C" + test.__name__) - class PyTest(test): - open = staticmethod(pyio.open) - PyTest.__name__ = str("Py" + test.__name__) - tests.append(CTest) - tests.append(PyTest) - support.run_unittest(*tests) + test_support.run_unittest( + TestNativeNewlines, + TestCRNewlines, + TestLFNewlines, + TestCRLFNewlines, + TestMixedNewlines + ) if __name__ == '__main__': test_main() -- 2.11.4.GIT