patch up leaking fds

This commit is contained in:
Benjamin Peterson 2010-10-31 01:30:11 +00:00
parent d8fc2e1aeb
commit ebe5d8ae3b
1 changed files with 21 additions and 22 deletions

View File

@ -555,12 +555,13 @@ def test_closefd_attr(self):
def test_garbage_collection(self): def test_garbage_collection(self):
# FileIO objects are collected, and collecting them flushes # FileIO objects are collected, and collecting them flushes
# all data to disk. # all data to disk.
f = self.FileIO(support.TESTFN, "wb") with support.check_warnings(('', ResourceWarning)):
f.write(b"abcxxx") f = self.FileIO(support.TESTFN, "wb")
f.f = f f.write(b"abcxxx")
wr = weakref.ref(f) f.f = f
del f wr = weakref.ref(f)
support.gc_collect() del f
support.gc_collect()
self.assertTrue(wr() is None, wr) self.assertTrue(wr() is None, wr)
with self.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")
@ -1984,26 +1985,24 @@ def test_seeking(self):
u_suffix = "\u8888\n" u_suffix = "\u8888\n"
suffix = bytes(u_suffix.encode("utf-8")) suffix = bytes(u_suffix.encode("utf-8"))
line = prefix + suffix line = prefix + suffix
f = self.open(support.TESTFN, "wb") with self.open(support.TESTFN, "wb") as f:
f.write(line*2) f.write(line*2)
f.close() with self.open(support.TESTFN, "r", encoding="utf-8") as f:
f = self.open(support.TESTFN, "r", encoding="utf-8") s = f.read(prefix_size)
s = f.read(prefix_size) self.assertEquals(s, str(prefix, "ascii"))
self.assertEquals(s, str(prefix, "ascii")) self.assertEquals(f.tell(), prefix_size)
self.assertEquals(f.tell(), prefix_size) self.assertEquals(f.readline(), u_suffix)
self.assertEquals(f.readline(), u_suffix)
def test_seeking_too(self): def test_seeking_too(self):
# Regression test for a specific bug # Regression test for a specific bug
data = b'\xe0\xbf\xbf\n' data = b'\xe0\xbf\xbf\n'
f = self.open(support.TESTFN, "wb") with self.open(support.TESTFN, "wb") as f:
f.write(data) f.write(data)
f.close() with self.open(support.TESTFN, "r", encoding="utf-8") as f:
f = self.open(support.TESTFN, "r", encoding="utf-8") f._CHUNK_SIZE # Just test that it exists
f._CHUNK_SIZE # Just test that it exists f._CHUNK_SIZE = 2
f._CHUNK_SIZE = 2 f.readline()
f.readline() f.tell()
f.tell()
def test_seek_and_tell(self): def test_seek_and_tell(self):
#Test seek/tell using the StatefulIncrementalDecoder. #Test seek/tell using the StatefulIncrementalDecoder.