mirror of
https://github.com/python/cpython.git
synced 2025-08-01 07:33:08 +00:00
Issue #6215: Fixed to use self.open() instead of open() or io.open().
This commit is contained in:
parent
5ecf57adcf
commit
cdcd4bff12
1 changed files with 18 additions and 18 deletions
|
@ -298,13 +298,13 @@ class IOTest(unittest.TestCase):
|
||||||
def test_invalid_operations(self):
|
def test_invalid_operations(self):
|
||||||
# Try writing on a file opened in read mode and vice-versa.
|
# Try writing on a file opened in read mode and vice-versa.
|
||||||
for mode in ("w", "wb"):
|
for mode in ("w", "wb"):
|
||||||
with open(support.TESTFN, mode) as fp:
|
with self.open(support.TESTFN, mode) as fp:
|
||||||
self.assertRaises(IOError, fp.read)
|
self.assertRaises(IOError, fp.read)
|
||||||
self.assertRaises(IOError, fp.readline)
|
self.assertRaises(IOError, fp.readline)
|
||||||
with open(support.TESTFN, "rb") as fp:
|
with self.open(support.TESTFN, "rb") as fp:
|
||||||
self.assertRaises(IOError, fp.write, b"blah")
|
self.assertRaises(IOError, fp.write, b"blah")
|
||||||
self.assertRaises(IOError, fp.writelines, [b"blah\n"])
|
self.assertRaises(IOError, fp.writelines, [b"blah\n"])
|
||||||
with open(support.TESTFN, "r") as fp:
|
with self.open(support.TESTFN, "r") as fp:
|
||||||
self.assertRaises(IOError, fp.write, "blah")
|
self.assertRaises(IOError, fp.write, "blah")
|
||||||
self.assertRaises(IOError, fp.writelines, ["blah\n"])
|
self.assertRaises(IOError, fp.writelines, ["blah\n"])
|
||||||
|
|
||||||
|
@ -375,12 +375,12 @@ class IOTest(unittest.TestCase):
|
||||||
def test_with_open(self):
|
def test_with_open(self):
|
||||||
for bufsize in (0, 1, 100):
|
for bufsize in (0, 1, 100):
|
||||||
f = None
|
f = None
|
||||||
with open(support.TESTFN, "wb", bufsize) as f:
|
with self.open(support.TESTFN, "wb", bufsize) as f:
|
||||||
f.write(b"xxx")
|
f.write(b"xxx")
|
||||||
self.assertEqual(f.closed, True)
|
self.assertEqual(f.closed, True)
|
||||||
f = None
|
f = None
|
||||||
try:
|
try:
|
||||||
with open(support.TESTFN, "wb", bufsize) as f:
|
with self.open(support.TESTFN, "wb", bufsize) as f:
|
||||||
1/0
|
1/0
|
||||||
except ZeroDivisionError:
|
except ZeroDivisionError:
|
||||||
self.assertEqual(f.closed, True)
|
self.assertEqual(f.closed, True)
|
||||||
|
@ -420,7 +420,7 @@ class IOTest(unittest.TestCase):
|
||||||
del f
|
del f
|
||||||
support.gc_collect()
|
support.gc_collect()
|
||||||
self.assertEqual(record, [1, 2, 3])
|
self.assertEqual(record, [1, 2, 3])
|
||||||
with open(support.TESTFN, "rb") as f:
|
with self.open(support.TESTFN, "rb") as f:
|
||||||
self.assertEqual(f.read(), b"xxx")
|
self.assertEqual(f.read(), b"xxx")
|
||||||
|
|
||||||
def _check_base_destructor(self, base):
|
def _check_base_destructor(self, base):
|
||||||
|
@ -515,7 +515,7 @@ class IOTest(unittest.TestCase):
|
||||||
del f
|
del f
|
||||||
support.gc_collect()
|
support.gc_collect()
|
||||||
self.assert_(wr() is None, wr)
|
self.assert_(wr() is None, wr)
|
||||||
with open(support.TESTFN, "rb") as f:
|
with self.open(support.TESTFN, "rb") as f:
|
||||||
self.assertEqual(f.read(), b"abcxxx")
|
self.assertEqual(f.read(), b"abcxxx")
|
||||||
|
|
||||||
def test_unbounded_file(self):
|
def test_unbounded_file(self):
|
||||||
|
@ -527,11 +527,11 @@ class IOTest(unittest.TestCase):
|
||||||
self.skipTest("test can only run in a 32-bit address space")
|
self.skipTest("test can only run in a 32-bit address space")
|
||||||
if support.real_max_memuse < support._2G:
|
if support.real_max_memuse < support._2G:
|
||||||
self.skipTest("test requires at least 2GB of memory")
|
self.skipTest("test requires at least 2GB of memory")
|
||||||
with open(zero, "rb", buffering=0) as f:
|
with self.open(zero, "rb", buffering=0) as f:
|
||||||
self.assertRaises(OverflowError, f.read)
|
self.assertRaises(OverflowError, f.read)
|
||||||
with open(zero, "rb") as f:
|
with self.open(zero, "rb") as f:
|
||||||
self.assertRaises(OverflowError, f.read)
|
self.assertRaises(OverflowError, f.read)
|
||||||
with open(zero, "r") as f:
|
with self.open(zero, "r") as f:
|
||||||
self.assertRaises(OverflowError, f.read)
|
self.assertRaises(OverflowError, f.read)
|
||||||
|
|
||||||
class CIOTest(IOTest):
|
class CIOTest(IOTest):
|
||||||
|
@ -744,9 +744,9 @@ class BufferedReaderTest(unittest.TestCase, CommonBufferedTests):
|
||||||
l = list(range(256)) * N
|
l = list(range(256)) * N
|
||||||
random.shuffle(l)
|
random.shuffle(l)
|
||||||
s = bytes(bytearray(l))
|
s = bytes(bytearray(l))
|
||||||
with io.open(support.TESTFN, "wb") as f:
|
with self.open(support.TESTFN, "wb") as f:
|
||||||
f.write(s)
|
f.write(s)
|
||||||
with io.open(support.TESTFN, self.read_mode, buffering=0) as raw:
|
with self.open(support.TESTFN, self.read_mode, buffering=0) as raw:
|
||||||
bufio = self.tp(raw, 8)
|
bufio = self.tp(raw, 8)
|
||||||
errors = []
|
errors = []
|
||||||
results = []
|
results = []
|
||||||
|
@ -974,12 +974,12 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
|
||||||
|
|
||||||
def test_truncate(self):
|
def test_truncate(self):
|
||||||
# Truncate implicitly flushes the buffer.
|
# Truncate implicitly flushes the buffer.
|
||||||
with io.open(support.TESTFN, self.write_mode, buffering=0) as raw:
|
with self.open(support.TESTFN, self.write_mode, buffering=0) as raw:
|
||||||
bufio = self.tp(raw, 8)
|
bufio = self.tp(raw, 8)
|
||||||
bufio.write(b"abcdef")
|
bufio.write(b"abcdef")
|
||||||
self.assertEqual(bufio.truncate(3), 3)
|
self.assertEqual(bufio.truncate(3), 3)
|
||||||
self.assertEqual(bufio.tell(), 3)
|
self.assertEqual(bufio.tell(), 3)
|
||||||
with io.open(support.TESTFN, "rb", buffering=0) as f:
|
with self.open(support.TESTFN, "rb", buffering=0) as f:
|
||||||
self.assertEqual(f.read(), b"abc")
|
self.assertEqual(f.read(), b"abc")
|
||||||
|
|
||||||
def test_threads(self):
|
def test_threads(self):
|
||||||
|
@ -1001,7 +1001,7 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
|
||||||
# writing the buffer to the raw streams. This is in addition
|
# writing the buffer to the raw streams. This is in addition
|
||||||
# to concurrency issues due to switching threads in the middle
|
# to concurrency issues due to switching threads in the middle
|
||||||
# of Python code.
|
# of Python code.
|
||||||
with io.open(support.TESTFN, self.write_mode, buffering=0) as raw:
|
with self.open(support.TESTFN, self.write_mode, buffering=0) as raw:
|
||||||
bufio = self.tp(raw, 8)
|
bufio = self.tp(raw, 8)
|
||||||
errors = []
|
errors = []
|
||||||
def f():
|
def f():
|
||||||
|
@ -1024,7 +1024,7 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
|
||||||
self.assertFalse(errors,
|
self.assertFalse(errors,
|
||||||
"the following exceptions were caught: %r" % errors)
|
"the following exceptions were caught: %r" % errors)
|
||||||
bufio.close()
|
bufio.close()
|
||||||
with io.open(support.TESTFN, "rb") as f:
|
with self.open(support.TESTFN, "rb") as f:
|
||||||
s = f.read()
|
s = f.read()
|
||||||
for i in range(256):
|
for i in range(256):
|
||||||
self.assertEquals(s.count(bytes([i])), N)
|
self.assertEquals(s.count(bytes([i])), N)
|
||||||
|
@ -1084,7 +1084,7 @@ class CBufferedWriterTest(BufferedWriterTest):
|
||||||
del f
|
del f
|
||||||
support.gc_collect()
|
support.gc_collect()
|
||||||
self.assert_(wr() is None, wr)
|
self.assert_(wr() is None, wr)
|
||||||
with open(support.TESTFN, "rb") as f:
|
with self.open(support.TESTFN, "rb") as f:
|
||||||
self.assertEqual(f.read(), b"123xxx")
|
self.assertEqual(f.read(), b"123xxx")
|
||||||
|
|
||||||
|
|
||||||
|
@ -2065,7 +2065,7 @@ class CTextIOWrapperTest(TextIOWrapperTest):
|
||||||
del t
|
del t
|
||||||
support.gc_collect()
|
support.gc_collect()
|
||||||
self.assert_(wr() is None, wr)
|
self.assert_(wr() is None, wr)
|
||||||
with open(support.TESTFN, "rb") as f:
|
with self.open(support.TESTFN, "rb") as f:
|
||||||
self.assertEqual(f.read(), b"456def")
|
self.assertEqual(f.read(), b"456def")
|
||||||
|
|
||||||
class PyTextIOWrapperTest(TextIOWrapperTest):
|
class PyTextIOWrapperTest(TextIOWrapperTest):
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue