Update of /cvsroot/pywin32/pywin32/win32/test
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv15608/test
Modified Files:
test_win32file.py
Log Message:
Fix a bug regarding reference counting and OVERLAPPED objects, as
reported on python-win32. Also borrowed the repro for the test suite.
Index: test_win32file.py
===================================================================
RCS file: /cvsroot/pywin32/pywin32/win32/test/test_win32file.py,v
retrieving revision 1.6
retrieving revision 1.7
diff -C2 -d -r1.6 -r1.7
*** test_win32file.py 12 Apr 2005 02:56:45 -0000 1.6
--- test_win32file.py 2 Jun 2005 00:03:34 -0000 1.7
***************
*** 1,9 ****
import unittest
! import win32api, win32file, win32con, pywintypes, winerror
import sys
import os
import tempfile
import sets
! import shutil
class TestSimpleOps(unittest.TestCase):
--- 1,10 ----
import unittest
! import win32api, win32file, win32pipe, win32con, pywintypes, winerror
import sys
import os
import tempfile
import sets
! import threading
! import time
class TestSimpleOps(unittest.TestCase):
***************
*** 177,180 ****
--- 178,238 ----
self.failUnlessEqual(hr, winerror.ERROR_INVALID_HANDLE)
+ def _IOCPServerThread(self, handle, port, drop_overlapped_reference):
+ overlapped = pywintypes.OVERLAPPED()
+ win32pipe.ConnectNamedPipe(handle, overlapped)
+ if drop_overlapped_reference:
+ # Be naughty - the overlapped object is now dead, but
+ # GetQueuedCompletionStatus will still find it. Our check of
+ # reference counting should catch that error.
+ overlapped = None
+ self.failUnlessRaises(RuntimeError,
+ win32file.GetQueuedCompletionStatus, port, -1)
+ handle.Close()
+ return
+
+ result = win32file.GetQueuedCompletionStatus(port, -1)
+ ol2 = result[-1]
+ self.failUnless(ol2 is overlapped)
+ data = win32file.ReadFile(handle, 512)[1]
+ win32file.WriteFile(handle, data)
+
+ def testCompletionPortsNonQueued(self, test_overlapped_death = 0):
+ # In 204 we had a reference count bug when OVERLAPPED objects were
+ # associated with a completion port other than via
+ # PostQueuedCompletionStatus. This test is based on the reproduction
+ # reported with that bug.
+ # Create the pipe.
+ BUFSIZE = 512
+ pipe_name = r"\\.\pipe\pywin32_test_pipe"
+ handle = win32pipe.CreateNamedPipe(pipe_name,
+ win32pipe.PIPE_ACCESS_DUPLEX|
+ win32file.FILE_FLAG_OVERLAPPED,
+ win32pipe.PIPE_TYPE_MESSAGE|
+ win32pipe.PIPE_READMODE_MESSAGE|
+ win32pipe.PIPE_WAIT,
+ 1, BUFSIZE, BUFSIZE,
+ win32pipe.NMPWAIT_WAIT_FOREVER,
+ None)
+ # Create an IOCP and associate it with the handle.
+ port = win32file.CreateIoCompletionPort(-1, 0, 0, 0)
+ win32file.CreateIoCompletionPort(handle, port, 1, 0)
+
+ thread = threading.Thread(target=self._IOCPServerThread, args=(handle,port, test_overlapped_death))
+ thread.start()
+ try:
+ time.sleep(0.1) # let thread do its thing.
+ try:
+ win32pipe.CallNamedPipe(r"\\.\pipe\pywin32_test_pipe", "Hello there", BUFSIZE, 0)
+ except win32pipe.error:
+ # Testing for overlapped death causes this
+ if not test_overlapped_death:
+ raise
+ finally:
+ handle.Close()
+ thread.join()
+
+ def testCompletionPortsNonQueuedBadReference(self):
+ self.testCompletionPortsNonQueued(True)
+
class TestFindFiles(unittest.TestCase):
***************
*** 198,202 ****
num = 0
for i in win32file.FindFilesIterator(spec):
- print "Got", i
num += 1
self.failUnlessEqual(0, num)
--- 256,259 ----
***************
*** 205,209 ****
test_path = os.path.join(win32api.GetTempPath(), "win32file_test_directory")
try:
! shutil.rmtree(test_path)
except os.error:
pass
--- 262,269 ----
test_path = os.path.join(win32api.GetTempPath(), "win32file_test_directory")
try:
! # Note: previously used shutil.rmtree, but when looking for
! # reference count leaks, that function showed leaks! os.rmdir
! # doesn't have that problem.
! os.rmdir(test_path)
except os.error:
pass
***************
*** 216,220 ****
self.failUnlessEqual(2, num)
finally:
! shutil.rmtree(test_path)
class TestEncrypt(unittest.TestCase):
--- 276,280 ----
self.failUnlessEqual(2, num)
finally:
! os.rmdir(test_path)
class TestEncrypt(unittest.TestCase):
|
